diff --git a/README.md b/README.md index 600f291..b21db48 100644 --- a/README.md +++ b/README.md @@ -1346,6 +1346,44 @@ existing = client.find_existing_components( `TangleApiClient` is handwritten in `tangle_cli.client` and inherits generated endpoint methods from `tangle_api.generated.operations.GeneratedTangleApiOperations`. The generated endpoint methods call the handwritten transport/request logic. Handwritten semantic helpers such as `find_existing_components(...)` return domain models and normalize common compatibility cases. +### Execution trees and request hooks + +`client.get_execution_details(execution_id)` returns the execution with its whole subtree attached (`child_executions`), and injects each child's execution context into the parent's graph tasks. This is what run details (`get_run_details`) and artifact lookups use. + +Since 0.1.34 the tree is fetched breadth-first by `tangle_cli.execution_tree.fetch_execution_tree`: +- one batch of `executions_details` requests per tree level; +- up to `client.execution_tree_max_workers` requests in parallel (default 8), run inline when threads are unavailable (`sys.platform == "emscripten"`) or with 1 worker; +- each execution is fetched once. + +The enriched result is identical to the previous one-request-per-node recursion; golden tests pin it. Downstream tools can call `fetch_execution_tree(client, root_id)` directly. It returns `{execution_id: TreeNode(execution_id, name, parent, details)}` in breadth-first order. + +Batched readers call two optional hooks on the client with relative API paths (for example `/api/executions/{id}/details`). Both are no-ops by default; override them in a subclass or wrapper to add caching: + +| Hook | When it's called | What an override can do | +|------|------------------|-------------------------| +| `prefetch(paths)` | Before a batch is read | Fetch the paths concurrently into a cache that its transport then serves | +| `persist(paths)` | After a batch of responses that never change, such as an execution's details | Keep those responses in a durable cache and serve later reads from it | + +```python +class CachingClient(TangleApiClient): + def prefetch(self, paths): + my_cache.warm(paths) # e.g. parallel GETs into a cache the transport reads + + def persist(self, paths): + my_cache.mark_immutable(paths) +``` + +### Client-side rate limiting + +Since 0.1.34 every client paces its own requests, so parallel readers such as the execution-tree fetch stay under Tangle's limits (about 10 requests/s per route per user): + +- **Pacing:** `TangleApiClient(..., max_requests_per_second=8.0)` spaces the start of every physical request at least `1/max_requests_per_second` apart. That includes retries and redirect hops, from every thread sharing the client. The default is `TangleApiClient.DEFAULT_MAX_REQUESTS_PER_SECOND` (8). `None` or `0` disables pacing. It can be changed on a live client (`client.max_requests_per_second = 4`). +- **How waiting works:** a caller takes a slot only when it can start right away. Otherwise it sleeps outside the lock and checks again, so nothing busy-waits, and a pause that starts while it's asleep still holds it. A single thread (inline mode) just sleeps between its own requests. Idle time is not saved up into a later burst. Changing the rate on a live client keeps any active pause. +- **Concurrency:** `execution_tree_max_workers` still caps parallel requests separately. Workers share one client: each `_refresh_auth()` and each per-send header snapshot runs under a per-client re-entrant lock. A custom `session=` must be safe to share across threads (`requests.Session` is, for ordinary use); otherwise set `execution_tree_max_workers = 1`. +- **Retry deadline:** waiting for the *first* send's slot happens after the auth refresh and before a request's retry budget starts, because queueing behind other workers isn't retrying. Waits for retries and redirect hops do count. If their next slot (pacing, or a shared pause) would only open after the retry deadline, the client gives up at once instead of sleeping past it. +- **Server pushback:** a `429` (`Retry-After`, or exponential backoff) or a `503` (its `Retry-After` is now honoured too, capped at 60 s like a 429's, including when opening a log stream) still retries within the existing shared retry budget. The retrying request's own wait is unchanged. While it waits, the client's shared limiter is paused for the same window. Other workers' next requests wait for the pause plus a small random jitter, up to 20% of the pause and at most 2 s, so they don't all resume at once. Other 5xx responses keep their per-request backoff without pausing anyone. +- **Pyodide:** under `sys.platform == "emscripten"` the default is `None`. A browser host such as tangle-js paces requests in its own transport, and responses its cache serves never reach the network, so pacing there too would only compound delays. A host that wants client-side pacing can pass a value explicitly. `prefetch`/`persist` hooks are not rate limited by the client; whatever transport serves those prefetched paths applies its own pacing. + The top-level `import tangle_cli` is lightweight and does not import static bindings eagerly. Normal installs include `tangle-api`; source checkouts or downstream embeddings may instead provide a local `tangle_api.generated` package before importing `tangle_cli.client`. ## Codegen/autogen from OpenAPI diff --git a/packages/tangle-cli/src/tangle_cli/__init__.py b/packages/tangle-cli/src/tangle_cli/__init__.py index c7e0852..b126e7d 100644 --- a/packages/tangle-cli/src/tangle_cli/__init__.py +++ b/packages/tangle-cli/src/tangle_cli/__init__.py @@ -14,6 +14,6 @@ try: __version__ = metadata_version("tangle-cli") except PackageNotFoundError: - __version__ = "0.1.33" + __version__ = "0.1.34" __all__ = ["TangleDynamicDiscoveryClient", "__version__"] diff --git a/packages/tangle-cli/src/tangle_cli/client.py b/packages/tangle-cli/src/tangle_cli/client.py index 5f07831..0d7631b 100644 --- a/packages/tangle-cli/src/tangle_cli/client.py +++ b/packages/tangle-cli/src/tangle_cli/client.py @@ -8,6 +8,8 @@ from __future__ import annotations +import sys +import threading import time from collections.abc import Iterable, Iterator, Mapping from dataclasses import asdict, is_dataclass @@ -32,6 +34,7 @@ from tangle_api.generated.operations import GeneratedTangleApiOperations from . import models as _cli_models from .logger import Logger, _null_logger, get_default_logger +from .rate_limit import RequestRateLimiter from .models import ( ComponentInfo, ComponentSpec, @@ -58,12 +61,15 @@ class _RetryBudget: further send is admitted. """ - __slots__ = ("attempts", "deadline", "sends") + __slots__ = ("attempts", "deadline", "first_slot_taken", "sends") def __init__(self, max_attempts: int, max_sends: int, deadline: float) -> None: self.attempts = max_attempts self.sends = max_sends self.deadline = deadline + # The first send's rate-limiter slot is taken before the budget exists + # (queueing behind other workers is not retrying), so it is not re-taken. + self.first_slot_taken = False def start_attempt(self) -> None: """Charge one logical try; callers gate on :meth:`can_retry` first.""" @@ -106,6 +112,9 @@ def allows_wait(self, delay: float) -> bool: return delay < self.deadline - time.monotonic() +_RATE_LIMITER_INIT_LOCK = threading.Lock() + + class TangleApiClient(GeneratedTangleApiOperations): """Single public API wrapper for Tangle backends. @@ -155,6 +164,14 @@ class TangleApiClient(GeneratedTangleApiOperations): # below attempts x chain length. _MAX_PHYSICAL_SENDS = max(2 * (_MAX_REDIRECTS + 1), 2 * (_MAX_GET_RETRIES + 1)) _MAX_RETRY_ELAPSED_SECONDS = 120.0 + #: Parallel ``executions_details`` requests per tree level in + #: :meth:`get_execution_details` (requests run inline without threads). + execution_tree_max_workers = 8 + #: Client-side pacing of every physical request (all threads sharing this + #: client), below Tangle's ~10 requests/s per route per user. ``None`` or + #: ``0`` disables pacing. Off by default under Pyodide (``emscripten``), + #: whose host transport paces requests itself and where sleeping blocks. + DEFAULT_MAX_REQUESTS_PER_SECOND: float | None = None if sys.platform == "emscripten" else 8.0 def __init__( self, @@ -170,8 +187,10 @@ def __init__( session: requests.Session | None = None, include_env_credentials: bool = True, verify: VerifyArgument = None, + max_requests_per_second: float | None = DEFAULT_MAX_REQUESTS_PER_SECOND, ) -> None: self.base_url = _normalize_base_url(base_url or default_base_url()) + self.max_requests_per_second = max_requests_per_second env_verbose = tangle_verbose_enabled() self.verbose = verbose or env_verbose self.logger = logger or (get_default_logger() if self.verbose else _null_logger) @@ -189,6 +208,70 @@ def _response_model(self, model_name: str, default: Any) -> Any: return getattr(_cli_models, model_name, default) + # ---- Optional request hooks (extension point) ----------------------- + + def prefetch(self, paths: list[str]) -> None: + """Hook: *paths* (relative API paths, e.g. ``/api/executions/{id}/details``) are about to be read. + + Batched readers such as :meth:`get_execution_details` call this before + issuing a batch of GETs, so a subclass or wrapper can fetch them + concurrently into a cache that its transport then serves. The default + does nothing. + """ + + def persist(self, paths: list[str]) -> None: + """Hook: the responses just read for *paths* are immutable and may be cached durably. + + Called after a batch of GETs whose responses never change (for example + an execution's details). A subclass or wrapper may keep them in a + durable cache and serve later reads from it. The default does nothing. + """ + + # ---- Shared client state --------------------------------------------- + + def _client_state_lock(self) -> Any: + """Re-entrant lock serializing auth refresh and per-send header snapshots. + + Parallel readers (e.g. :meth:`get_execution_details`) share one client; + ``_refresh_auth`` overrides may mutate ``self.headers``/session state, so + each refresh and each header snapshot runs under this lock. Re-entrant, + so a refresh that itself issues requests through this client is safe. + The ``session`` passed to the constructor must be safe to share across + threads (``requests.Session`` is, for ordinary use); otherwise set + ``execution_tree_max_workers = 1``. + """ + + lock = self.__dict__.get("_state_lock") + if lock is None: + with _RATE_LIMITER_INIT_LOCK: + lock = self.__dict__.get("_state_lock") + if lock is None: + lock = threading.RLock() + self.__dict__["_state_lock"] = lock + return lock + + # ---- Client-side rate limiting --------------------------------------- + + def _request_rate_limiter(self) -> RequestRateLimiter: + """The limiter shared by every thread using this client (created lazily). + + A ``max_requests_per_second`` changed after construction updates the + existing limiter in place, so an active server-pushback pause is kept. + """ + + rps = getattr(self, "max_requests_per_second", self.DEFAULT_MAX_REQUESTS_PER_SECOND) + wanted = rps if rps and rps > 0 else None + limiter = self.__dict__.get("_rate_limiter") + if limiter is None: + with _RATE_LIMITER_INIT_LOCK: + limiter = self.__dict__.get("_rate_limiter") + if limiter is None: + limiter = RequestRateLimiter(wanted) + self.__dict__["_rate_limiter"] = limiter + if limiter.max_per_second != wanted: + limiter.set_rate(wanted) + return limiter + def set_verbose(self, enabled: bool) -> None: """Enable or disable request logging.""" @@ -225,13 +308,21 @@ def _make_request( clean_params = self._clean_mapping(params) request_method = method.upper() + with self._client_state_lock(): + self._refresh_auth() + # Queue for the first send's pacing slot AFTER the (serialized) auth + # refresh, so workers released together from a slow refresh are still + # paced, and BEFORE the retry budget starts: waiting behind other workers + # sharing this client is not retrying, so it must not spend the retry + # deadline. Later sends (retries, redirect hops) take their slot inside + # the budget, before its deadline check. + self._request_rate_limiter().acquire() budget = _RetryBudget( self._MAX_GET_RETRIES + 1, self._MAX_PHYSICAL_SENDS, time.monotonic() + self._MAX_RETRY_ELAPSED_SECONDS, ) - - self._refresh_auth() + budget.first_slot_taken = True response = self._request_with_rate_limit_retries( request_method, url, @@ -251,7 +342,8 @@ def _make_request( # leaking it. ``response.headers`` stays available after close. if kwargs.get("stream"): response.close() - self._refresh_auth() + with self._client_state_lock(): + self._refresh_auth() try: response = self._request_with_rate_limit_retries( request_method, @@ -328,6 +420,9 @@ def _request_with_rate_limit_retries( # still reportable if a later round exhausts the budget mid-chain. last_response = response self._release_response(response) + # Hold this client's other workers for the same window so they don't + # keep hitting the limit; this thread sleeps through it below. + self._request_rate_limiter().pause(delay) self._sleep_for_rate_limit(delay) rate_limit_round += 1 @@ -426,10 +521,18 @@ def _request_with_transient_retries( else: if response.status_code not in self._RETRYABLE_GET_STATUSES: return response + if response.status_code == 503: + # A 503 is the server shedding load: honour its Retry-After + # (capped like a 429's) and hold this client's other workers. + retry_after = self._retry_after_delay(response.headers.get("Retry-After")) + if retry_after is not None: + delay = max(delay, min(retry_after, self._MAX_RETRY_AFTER_SECONDS)) # Budget exhausted, or the backoff alone would outlast the # deadline: return the final 5xx for raise_for_status. if not budget.can_retry() or not budget.allows_wait(delay): return response + if response.status_code == 503: + self._request_rate_limiter().pause(delay) # Release the intermediate response so its connection returns to # the pool. Its body is already buffered (these GETs are never # streamed), so it stays reportable if the retry runs out of @@ -457,8 +560,12 @@ def _sleep_for_rate_limit(self, delay: float) -> None: self.logger.info(f"429 rate limited; retrying in {delay:.1f}s") time.sleep(delay) - def _sleep_for_stream_open_retry(self, backoff: float, next_attempt: int, reason: str) -> None: + def _sleep_for_stream_open_retry( + self, backoff: float, next_attempt: int, reason: str, *, minimum: float | None = None + ) -> None: delay = min(backoff, self._MAX_STREAM_OPEN_BACKOFF_SECONDS) + if minimum is not None: + delay = max(delay, minimum) # e.g. a 503's Retry-After self.logger.warn( f"transient {reason} opening log stream; retrying in {delay:.1f}s " f"(attempt {next_attempt}/{self._MAX_STREAM_OPEN_ATTEMPTS})" @@ -516,12 +623,27 @@ def _request_with_same_origin_redirects( response: requests.Response | None = None for _ in range(self._MAX_REDIRECTS + 1): + # Pace (and honour any shared 429/503 pause) BEFORE consuming the send: + # the budget's deadline check then sees the time spent waiting, so a + # limiter wait can never carry a send past the retry deadline. + remaining = budget.deadline - time.monotonic() + if budget.first_slot_taken: + budget.first_slot_taken = False + elif remaining > 0 and self._request_rate_limiter().acquire(max_wait=remaining) is None: + # The next slot (pacing, or a shared 429/503 pause) would only open + # after the retry deadline: give up now instead of sleeping past it. + # (An already-passed deadline is left to the budget check below.) + raise requests.exceptions.RetryError( + f"Retry budget exhausted (rate-limit wait would outlast the deadline) " + f"while sending {current_method} {self._credential_safe_url(current_url)}" + ) if not budget.try_consume_send(): raise requests.exceptions.RetryError( f"Retry budget exhausted ({budget.exhaustion_reason()}) " f"while sending {current_method} {self._credential_safe_url(current_url)}" ) - request_headers = self._headers(extra_headers) + with self._client_state_lock(): # consistent snapshot vs a concurrent _refresh_auth + request_headers = self._headers(extra_headers) call_kwargs = dict(request_kwargs) if self._verify is not _VERIFY_UNSET and "verify" not in call_kwargs: call_kwargs["verify"] = self._verify @@ -697,8 +819,21 @@ def _decode_response(response: requests.Response) -> Any: # ---- Handwritten semantic helpers consumed by tangle-deploy ---------- def get_execution_details(self, execution_id: str) -> GetExecutionInfoResponse: - details = self.executions_details(execution_id) - self._enrich_execution_tree(details) + """Fetch an execution and its whole subtree, enriched with execution context. + + The tree is fetched breadth-first with + :func:`tangle_cli.execution_tree.fetch_execution_tree`: one batch per + level, ``execution_tree_max_workers`` parallel requests (inline without + threads), and the :meth:`prefetch`/:meth:`persist` hooks around each + batch. :meth:`_enrich_execution_tree` then builds the same enriched + tree as the historical one-request-per-node recursion. + """ + + from .execution_tree import fetch_execution_tree + + tree = fetch_execution_tree(self, execution_id, max_workers=self.execution_tree_max_workers) + details = tree[execution_id].details + self._enrich_execution_tree(details, {eid: node.details for eid, node in tree.items()}) return details @staticmethod @@ -787,8 +922,18 @@ def stream_execution_container_log(self, execution_id: str) -> requests.Response last_exc = None if attempt == self._MAX_STREAM_OPEN_ATTEMPTS: break + minimum = None + if response.status_code == 503: + # Load shedding: honour Retry-After (capped like a 429's) and + # hold this client's other workers for the same window. + retry_after = self._retry_after_delay(response.headers.get("Retry-After")) + if retry_after is not None: + minimum = min(retry_after, self._MAX_RETRY_AFTER_SECONDS) + self._request_rate_limiter().pause( + max(min(backoff, self._MAX_STREAM_OPEN_BACKOFF_SECONDS), minimum or 0.0) + ) self._sleep_for_stream_open_retry( - backoff, attempt + 1, f"HTTP {response.status_code}" + backoff, attempt + 1, f"HTTP {response.status_code}", minimum=minimum ) backoff *= 2.0 continue @@ -1052,7 +1197,18 @@ def get_run_pipeline_spec(self, run_id: str) -> TaskSpec | None: execution = self.executions_details(root_execution_id) return execution.task_spec - def _enrich_execution_tree(self, execution: GetExecutionInfoResponse) -> None: + def _enrich_execution_tree( + self, + execution: GetExecutionInfoResponse, + fetched: Mapping[str, GetExecutionInfoResponse] | None = None, + ) -> None: + """Attach child executions and inject execution context into graph tasks, recursively. + + *fetched* maps execution ids to already-fetched details (see + :meth:`get_execution_details`); a child missing from it is fetched + with ``executions_details``. + """ + child_ids = execution.raw.get("child_task_execution_ids") or {} if not isinstance(child_ids, dict): return @@ -1061,8 +1217,10 @@ def _enrich_execution_tree(self, execution: GetExecutionInfoResponse) -> None: for task_name, child_execution_id in child_ids.items(): if not child_execution_id: continue - child = self.executions_details(child_execution_id) - self._enrich_execution_tree(child) + child = fetched.get(child_execution_id) if fetched is not None else None + if child is None: + child = self.executions_details(child_execution_id) + self._enrich_execution_tree(child, fetched) execution.child_executions[task_name] = child task = execution.task_spec.graph_tasks.get(task_name) diff --git a/packages/tangle-cli/src/tangle_cli/execution_tree.py b/packages/tangle-cli/src/tangle_cli/execution_tree.py new file mode 100644 index 0000000..72a4280 --- /dev/null +++ b/packages/tangle-cli/src/tangle_cli/execution_tree.py @@ -0,0 +1,138 @@ +"""Breadth-first, batched fetching of a run's execution tree. + +:func:`fetch_execution_tree` fetches ``/api/executions/{id}/details`` for an +execution and everything under it, one batch per tree level: + +* on CPython each batch is fetched with a small thread pool; without threads + (e.g. Pyodide / ``sys.platform == "emscripten"``) or with ``max_workers=1`` + it runs inline; +* the client's optional ``prefetch(paths)`` / ``persist(paths)`` hooks (see + :meth:`tangle_cli.client.TangleApiClient.prefetch`) are called with each + batch's relative API paths: ``prefetch`` before the batch is read, ``persist`` + after it (an execution's details never change). + +:meth:`tangle_cli.client.TangleApiClient.get_execution_details` uses it, and +downstream tools (e.g. run timelines) can reuse it directly. + +Importing this module starts no threads. +""" + +from __future__ import annotations + +import sys +from collections.abc import Callable +from dataclasses import dataclass +from typing import Any + +DEFAULT_MAX_WORKERS = 8 + + +def parallel_map(fn: Callable[[Any], Any], items: list[Any], max_workers: int) -> list[Any]: + """Map *fn* over *items* preserving order; inline without threads or for 1 worker. + + Exceptions raised by *fn* propagate to the caller. + """ + + if max_workers <= 1 or len(items) <= 1 or sys.platform == "emscripten": + return [fn(item) for item in items] + from concurrent.futures import ThreadPoolExecutor + + with ThreadPoolExecutor(max_workers=min(max_workers, len(items))) as pool: + return list(pool.map(fn, items)) + + +def call_client_hook(client: Any, hook: str, paths: list[str]) -> None: + """Call the client's optional ``prefetch``/``persist`` hook with relative API paths.""" + + function = getattr(client, hook, None) + if callable(function) and paths: + function(paths) + + +def details_path(execution_id: str) -> str: + """Relative API path of one execution's details (the path ``executions_details`` requests).""" + + return f"/api/executions/{execution_id}/details" + + +def child_execution_ids(details: Any) -> dict[str, str]: + """``{task name: child execution id}`` of one details response, in task order. + + Accepts a dict or a response model; reads the model's ``raw`` payload first, + like ``TangleApiClient._enrich_execution_tree``. Empty child ids are skipped. + """ + + if isinstance(details, dict): + children = details.get("child_task_execution_ids") + else: + raw = getattr(details, "raw", None) + children = raw.get("child_task_execution_ids") if isinstance(raw, dict) else None + if children is None: + children = getattr(details, "child_task_execution_ids", None) + if not isinstance(children, dict): + return {} + return {str(name): str(child) for name, child in children.items() if child} + + +@dataclass +class TreeNode: + """One fetched execution and its place in the tree.""" + + execution_id: str + name: str | None # task name in the parent graph; None for the root + parent: str | None + details: Any # the ``executions_details`` response (model or dict) + + +def fetch_execution_tree( + client: Any, + root_execution_id: str, + *, + fetch: Callable[[str], Any] | None = None, + max_workers: int = DEFAULT_MAX_WORKERS, +) -> dict[str, TreeNode]: + """Fetch every execution under (and including) *root_execution_id*, breadth-first. + + Args: + client: Supplies ``executions_details`` and the optional + ``prefetch``/``persist`` hooks. + root_execution_id: The tree (or subtree) root. + fetch: Override for fetching one execution's details (defaults to + ``client.executions_details``), e.g. to count or wrap calls. + max_workers: Parallel requests per level (inline without threads). + + Returns: + ``{execution_id: TreeNode}`` in breadth-first order, children in task + order. Each execution is fetched once. Exceptions from a fetch propagate. + """ + + fetch_one = fetch or client.executions_details + nodes: dict[str, TreeNode] = {} + frontier: list[tuple[str, str | None, str | None]] = [(root_execution_id, None, None)] + while frontier: + frontier = [entry for entry in frontier if entry[0] not in nodes] + ids = [execution_id for execution_id, _, _ in frontier] + if not ids: + break + call_client_hook(client, "prefetch", [details_path(execution_id) for execution_id in ids]) + responses = parallel_map(fetch_one, ids, max_workers) + next_frontier: list[tuple[str, str | None, str | None]] = [] + for (execution_id, name, parent), details in zip(frontier, responses): + nodes[execution_id] = TreeNode(execution_id=execution_id, name=name, parent=parent, details=details) + next_frontier.extend( + (child, child_name, execution_id) for child_name, child in child_execution_ids(details).items() + ) + call_client_hook(client, "persist", [details_path(execution_id) for execution_id in ids]) + frontier = next_frontier + return nodes + + +__all__ = [ + "DEFAULT_MAX_WORKERS", + "TreeNode", + "call_client_hook", + "child_execution_ids", + "details_path", + "fetch_execution_tree", + "parallel_map", +] diff --git a/packages/tangle-cli/src/tangle_cli/rate_limit.py b/packages/tangle-cli/src/tangle_cli/rate_limit.py new file mode 100644 index 0000000..637be17 --- /dev/null +++ b/packages/tangle-cli/src/tangle_cli/rate_limit.py @@ -0,0 +1,119 @@ +"""Client-side request pacing shared by every thread using one API client. + +:class:`RequestRateLimiter` spaces request *starts* at least ``1 / max_per_second`` +apart. A caller takes a slot under a lock only when it can start *now*; otherwise it +sleeps outside the lock until its earliest possible start and checks again. Callers +never busy-wait, a single thread (inline / no-threads mode) simply sleeps between +its own requests, and a pause that begins while a caller is asleep still holds it. + +:meth:`RequestRateLimiter.pause` is called when the server pushes back (429, or a +503 with ``Retry-After``). Until the pause ends, every *other* caller's next slot +waits for it, plus a small random jitter, so parallel workers don't all resume at +the same instant. The pausing caller is exempt: it already sleeps through its own +retry backoff. + +Pacing can be disabled (``max_per_second`` of ``None`` or ``0``). Pauses still +apply then. The clock, sleep and randomness are injectable so tests are +deterministic. +""" + +from __future__ import annotations + +import random +import threading +import time +from collections.abc import Callable + +# Bound at import so tests that monkeypatch ``time.sleep`` / ``time.monotonic`` for +# retry assertions don't also capture (or fake) the limiter's own waiting. +_MONOTONIC = time.monotonic +_SLEEP = time.sleep + +#: Jitter added to a pause's end for waiting callers: a fraction of the pause, capped. +PAUSE_JITTER_FRACTION = 0.2 +PAUSE_JITTER_MAX_SECONDS = 2.0 + + +class RequestRateLimiter: + """Thread-safe minimum-interval limiter with a shared server-pushback pause.""" + + def __init__( + self, + max_per_second: float | None, + *, + clock: Callable[[], float] | None = None, + sleep: Callable[[float], None] | None = None, + rand: Callable[[], float] | None = None, + thread_id: Callable[[], int] | None = None, + ) -> None: + self._interval = self._interval_for(max_per_second) + self._clock = clock or _MONOTONIC + self._sleep = sleep or _SLEEP + self._rand = rand or random.random + self._thread_id = thread_id or threading.get_ident + self._lock = threading.Lock() + self._last_start: float | None = None # next slot = last start + current interval + self._paused_until = 0.0 + self._pause_owner: int | None = None + + @staticmethod + def _interval_for(max_per_second: float | None) -> float: + return 1.0 / max_per_second if max_per_second and max_per_second > 0 else 0.0 + + @property + def max_per_second(self) -> float | None: + return 1.0 / self._interval if self._interval else None + + def set_rate(self, max_per_second: float | None) -> None: + """Change the pacing rate in place; an active pause and its owner are kept. + + The next slot is derived from the last start and the *new* interval, so + lowering the rate takes effect for the very next request. + """ + + with self._lock: + self._interval = self._interval_for(max_per_second) + + def acquire(self, max_wait: float | None = None) -> float | None: + """Wait for this caller's request slot; return the seconds waited. + + The slot is taken only once the caller can start immediately, after + re-checking any pause that began (or was extended) while it slept. With + *max_wait* (seconds, relative, so independent of the caller's clock), a + slot that cannot start within it in total is refused before sleeping: + ``None`` is returned and no slot is taken. + """ + + waited = 0.0 + while True: + with self._lock: + now = self._clock() + start = now + if self._interval and self._last_start is not None: + start = max(now, self._last_start + self._interval) + if self._paused_until > now and self._thread_id() != self._pause_owner: + remaining = self._paused_until - now + jitter = min(PAUSE_JITTER_MAX_SECONDS, PAUSE_JITTER_FRACTION * remaining) * self._rand() + start = max(start, self._paused_until + jitter) + if start <= now: + self._last_start = now + return waited + if max_wait is not None and waited + (start - now) > max_wait: + return None + wait = start - now + self._sleep(wait) + waited += wait + + def pause(self, seconds: float) -> None: + """Hold every other caller's requests for *seconds* (server asked us to back off).""" + + if seconds <= 0: + return + with self._lock: + until = self._clock() + seconds + if until > self._paused_until: + self._paused_until = until + self._pause_owner = self._thread_id() + + +__all__ = ["PAUSE_JITTER_FRACTION", "PAUSE_JITTER_MAX_SECONDS", "RequestRateLimiter"] diff --git a/pyproject.toml b/pyproject.toml index 4de5b27..56e27fd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "tangle-cli" -version = "0.1.33" +version = "0.1.34" description = "CLI for Tangle, the open-source ML pipeline orchestration platform" readme = "README.md" authors = [ diff --git a/tests/fixtures/run_details/compact.json b/tests/fixtures/run_details/compact.json new file mode 100644 index 0000000..afc25e5 --- /dev/null +++ b/tests/fixtures/run_details/compact.json @@ -0,0 +1,195 @@ +{ + "run": { + "id": "run-1", + "root_execution_id": "root-exec", + "created_at": "2026-06-11T00:00:00Z", + "created_by": "tester@example.com", + "annotations": { + "team": "search" + } + }, + "execution": { + "id": "root-exec", + "component": "Golden Pipeline", + "transparent": false, + "transparency_reason": "no inline source, canonical location, git metadata, or standard public image found", + "implementation": { + "graph": { + "tasks": { + "Prepare": { + "componentRef": { + "name": "Prepare", + "digest": "digest-Prepare", + "spec": { + "name": "Prepare", + "description": "Prepare does work", + "metadata": { + "annotations": { + "version": "1.0" + } + }, + "inputs": [ + { + "name": "in", + "type": "String" + } + ], + "outputs": [ + { + "name": "out", + "type": "String" + } + ] + } + }, + "arguments": { + "in": "hello" + }, + "execution_id": "prep-exec", + "input_artifacts": {}, + "output_artifacts": { + "out": "art-prep" + }, + "state": "SUCCEEDED" + }, + "Fan": { + "componentRef": { + "name": "Fan", + "digest": "digest-Fan", + "spec": { + "name": "Fan", + "implementation": { + "graph": { + "tasks": { + "Score A": { + "componentRef": { + "name": "Score", + "digest": "digest-Score", + "spec": { + "name": "Score", + "description": "Score does work", + "metadata": { + "annotations": { + "version": "1.0" + } + }, + "inputs": [ + { + "name": "in", + "type": "String" + } + ], + "outputs": [ + { + "name": "out", + "type": "String" + } + ] + } + }, + "arguments": { + "in": "hello" + }, + "execution_id": "score-a-exec", + "input_artifacts": { + "in": "art-prep" + }, + "output_artifacts": { + "out": "art-a" + }, + "state": "SUCCEEDED" + }, + "Score B": { + "componentRef": { + "name": "Score", + "digest": "digest-Score", + "spec": { + "name": "Score", + "description": "Score does work", + "metadata": { + "annotations": { + "version": "1.0" + } + }, + "inputs": [ + { + "name": "in", + "type": "String" + } + ], + "outputs": [ + { + "name": "out", + "type": "String" + } + ] + } + }, + "arguments": { + "in": "hello" + }, + "execution_id": "score-b-exec", + "input_artifacts": { + "in": "art-prep" + }, + "output_artifacts": { + "out": "art-b" + }, + "state": "FAILED" + } + } + } + } + } + }, + "arguments": {}, + "execution_id": "fan-exec", + "input_artifacts": { + "in": "art-prep" + }, + "output_artifacts": {} + }, + "Report": { + "componentRef": { + "name": "Report", + "digest": "digest-Report", + "spec": { + "name": "Report", + "description": "Report does work", + "metadata": { + "annotations": { + "version": "1.0" + } + }, + "inputs": [ + { + "name": "in", + "type": "String" + } + ], + "outputs": [ + { + "name": "out", + "type": "String" + } + ] + } + }, + "arguments": { + "in": "hello" + }, + "execution_id": "report-exec", + "input_artifacts": { + "a": "art-a", + "b": "art-b" + }, + "output_artifacts": {} + } + } + } + }, + "arguments": { + "arg": "value" + } + } +} diff --git a/tests/fixtures/run_details/implementations.json b/tests/fixtures/run_details/implementations.json new file mode 100644 index 0000000..8013fbb --- /dev/null +++ b/tests/fixtures/run_details/implementations.json @@ -0,0 +1,224 @@ +{ + "run": { + "id": "run-1", + "root_execution_id": "root-exec", + "created_at": "2026-06-11T00:00:00Z", + "created_by": "tester@example.com", + "annotations": { + "team": "search" + } + }, + "execution": { + "id": "root-exec", + "component": "Golden Pipeline", + "transparent": false, + "transparency_reason": "no inline source, canonical location, git metadata, or standard public image found", + "implementation": { + "graph": { + "tasks": { + "Prepare": { + "componentRef": { + "name": "Prepare", + "digest": "digest-Prepare", + "spec": { + "name": "Prepare", + "description": "Prepare does work", + "metadata": { + "annotations": { + "version": "1.0", + "python_dependencies": "[big list]" + } + }, + "inputs": [ + { + "name": "in", + "type": "String" + } + ], + "outputs": [ + { + "name": "out", + "type": "String" + } + ], + "implementation": { + "container": { + "image": "gcr.io/prep:v1" + } + } + }, + "text": "raw yaml for Prepare" + }, + "arguments": { + "in": "hello" + }, + "execution_id": "prep-exec", + "input_artifacts": {}, + "output_artifacts": { + "out": "art-prep" + }, + "state": "SUCCEEDED" + }, + "Fan": { + "componentRef": { + "name": "Fan", + "digest": "digest-Fan", + "spec": { + "name": "Fan", + "implementation": { + "graph": { + "tasks": { + "Score A": { + "componentRef": { + "name": "Score", + "digest": "digest-Score", + "spec": { + "name": "Score", + "description": "Score does work", + "metadata": { + "annotations": { + "version": "1.0", + "python_dependencies": "[big list]" + } + }, + "inputs": [ + { + "name": "in", + "type": "String" + } + ], + "outputs": [ + { + "name": "out", + "type": "String" + } + ], + "implementation": { + "container": { + "image": "gcr.io/score:v2" + } + } + }, + "text": "raw yaml for Score" + }, + "arguments": { + "in": "hello" + }, + "execution_id": "score-a-exec", + "input_artifacts": { + "in": "art-prep" + }, + "output_artifacts": { + "out": "art-a" + }, + "state": "SUCCEEDED" + }, + "Score B": { + "componentRef": { + "name": "Score", + "digest": "digest-Score", + "spec": { + "name": "Score", + "description": "Score does work", + "metadata": { + "annotations": { + "version": "1.0", + "python_dependencies": "[big list]" + } + }, + "inputs": [ + { + "name": "in", + "type": "String" + } + ], + "outputs": [ + { + "name": "out", + "type": "String" + } + ], + "implementation": { + "container": { + "image": "gcr.io/score:v2" + } + } + }, + "text": "raw yaml for Score" + }, + "arguments": { + "in": "hello" + }, + "execution_id": "score-b-exec", + "input_artifacts": { + "in": "art-prep" + }, + "output_artifacts": { + "out": "art-b" + }, + "state": "FAILED" + } + } + } + } + }, + "text": "raw yaml for Fan" + }, + "arguments": {}, + "execution_id": "fan-exec", + "input_artifacts": { + "in": "art-prep" + }, + "output_artifacts": {} + }, + "Report": { + "componentRef": { + "name": "Report", + "digest": "digest-Report", + "spec": { + "name": "Report", + "description": "Report does work", + "metadata": { + "annotations": { + "version": "1.0", + "python_dependencies": "[big list]" + } + }, + "inputs": [ + { + "name": "in", + "type": "String" + } + ], + "outputs": [ + { + "name": "out", + "type": "String" + } + ], + "implementation": { + "container": { + "image": "gcr.io/report:v1" + } + } + }, + "text": "raw yaml for Report" + }, + "arguments": { + "in": "hello" + }, + "execution_id": "report-exec", + "input_artifacts": { + "a": "art-a", + "b": "art-b" + }, + "output_artifacts": {} + } + } + } + }, + "arguments": { + "arg": "value" + } + } +} diff --git a/tests/fixtures/run_details/narrowed_to_subgraph.json b/tests/fixtures/run_details/narrowed_to_subgraph.json new file mode 100644 index 0000000..2a80cb8 --- /dev/null +++ b/tests/fixtures/run_details/narrowed_to_subgraph.json @@ -0,0 +1,105 @@ +{ + "run": { + "id": "run-1", + "root_execution_id": "root-exec", + "created_at": "2026-06-11T00:00:00Z", + "created_by": "tester@example.com", + "annotations": { + "team": "search" + } + }, + "execution": { + "id": "fan-exec", + "component": "Fan", + "transparent": false, + "transparency_reason": "no inline source, canonical location, git metadata, or standard public image found", + "implementation": { + "graph": { + "tasks": { + "Score A": { + "componentRef": { + "name": "Score", + "digest": "digest-Score", + "spec": { + "name": "Score", + "description": "Score does work", + "metadata": { + "annotations": { + "version": "1.0" + } + }, + "inputs": [ + { + "name": "in", + "type": "String" + } + ], + "outputs": [ + { + "name": "out", + "type": "String" + } + ] + } + }, + "arguments": { + "in": "hello" + }, + "execution_id": "score-a-exec", + "input_artifacts": { + "in": "art-prep" + }, + "output_artifacts": { + "out": "art-a" + }, + "state": "SUCCEEDED" + }, + "Score B": { + "componentRef": { + "name": "Score", + "digest": "digest-Score", + "spec": { + "name": "Score", + "description": "Score does work", + "metadata": { + "annotations": { + "version": "1.0" + } + }, + "inputs": [ + { + "name": "in", + "type": "String" + } + ], + "outputs": [ + { + "name": "out", + "type": "String" + } + ] + } + }, + "arguments": { + "in": "hello" + }, + "execution_id": "score-b-exec", + "input_artifacts": { + "in": "art-prep" + }, + "output_artifacts": { + "out": "art-b" + }, + "state": "FAILED" + } + } + } + }, + "arguments": { + "arg": "value" + }, + "input_artifacts": { + "in": "art-prep" + } + } +} diff --git a/tests/fixtures/run_details/state_and_annotations.json b/tests/fixtures/run_details/state_and_annotations.json new file mode 100644 index 0000000..e64cc25 --- /dev/null +++ b/tests/fixtures/run_details/state_and_annotations.json @@ -0,0 +1,214 @@ +{ + "run": { + "id": "run-1", + "root_execution_id": "root-exec", + "created_at": "2026-06-11T00:00:00Z", + "created_by": "tester@example.com", + "annotations": { + "team": "search" + } + }, + "execution": { + "id": "root-exec", + "component": "Golden Pipeline", + "transparent": false, + "transparency_reason": "no inline source, canonical location, git metadata, or standard public image found", + "implementation": { + "graph": { + "tasks": { + "Prepare": { + "componentRef": { + "name": "Prepare", + "digest": "digest-Prepare", + "spec": { + "name": "Prepare", + "description": "Prepare does work", + "metadata": { + "annotations": { + "version": "1.0" + } + }, + "inputs": [ + { + "name": "in", + "type": "String" + } + ], + "outputs": [ + { + "name": "out", + "type": "String" + } + ] + } + }, + "arguments": { + "in": "hello" + }, + "execution_id": "prep-exec", + "input_artifacts": {}, + "output_artifacts": { + "out": "art-prep" + }, + "state": "SUCCEEDED" + }, + "Fan": { + "componentRef": { + "name": "Fan", + "digest": "digest-Fan", + "spec": { + "name": "Fan", + "implementation": { + "graph": { + "tasks": { + "Score A": { + "componentRef": { + "name": "Score", + "digest": "digest-Score", + "spec": { + "name": "Score", + "description": "Score does work", + "metadata": { + "annotations": { + "version": "1.0" + } + }, + "inputs": [ + { + "name": "in", + "type": "String" + } + ], + "outputs": [ + { + "name": "out", + "type": "String" + } + ] + } + }, + "arguments": { + "in": "hello" + }, + "execution_id": "score-a-exec", + "input_artifacts": { + "in": "art-prep" + }, + "output_artifacts": { + "out": "art-a" + }, + "state": "SUCCEEDED" + }, + "Score B": { + "componentRef": { + "name": "Score", + "digest": "digest-Score", + "spec": { + "name": "Score", + "description": "Score does work", + "metadata": { + "annotations": { + "version": "1.0" + } + }, + "inputs": [ + { + "name": "in", + "type": "String" + } + ], + "outputs": [ + { + "name": "out", + "type": "String" + } + ] + } + }, + "arguments": { + "in": "hello" + }, + "execution_id": "score-b-exec", + "input_artifacts": { + "in": "art-prep" + }, + "output_artifacts": { + "out": "art-b" + }, + "state": "FAILED" + } + } + } + } + } + }, + "arguments": {}, + "execution_id": "fan-exec", + "input_artifacts": { + "in": "art-prep" + }, + "output_artifacts": {} + }, + "Report": { + "componentRef": { + "name": "Report", + "digest": "digest-Report", + "spec": { + "name": "Report", + "description": "Report does work", + "metadata": { + "annotations": { + "version": "1.0" + } + }, + "inputs": [ + { + "name": "in", + "type": "String" + } + ], + "outputs": [ + { + "name": "out", + "type": "String" + } + ] + } + }, + "arguments": { + "in": "hello" + }, + "execution_id": "report-exec", + "input_artifacts": { + "a": "art-a", + "b": "art-b" + }, + "output_artifacts": {} + } + } + } + }, + "arguments": { + "arg": "value" + } + }, + "annotations": { + "team": "search", + "owner": "tester" + }, + "execution_state": { + "totals": { + "SUCCEEDED": 2, + "FAILED": 1 + }, + "per_execution": { + "prep-exec": { + "SUCCEEDED": 1 + }, + "fan-exec": { + "SUCCEEDED": 1, + "FAILED": 1 + } + } + } +} diff --git a/tests/test_execution_details_golden.py b/tests/test_execution_details_golden.py new file mode 100644 index 0000000..a374805 --- /dev/null +++ b/tests/test_execution_details_golden.py @@ -0,0 +1,187 @@ +"""Regression pin for run-details output (``PipelineRunDetails.get_run_details_output``). + +The goldens in ``tests/fixtures/run_details`` were generated from the sequential +recursive traversal (one ``executions_details`` call per execution inside +``_enrich_execution_tree``), before ``get_execution_details`` fetched the tree with +``tangle_cli.execution_tree.fetch_execution_tree``. The output must stay +byte-identical however the tree is fetched. + +Regenerate (only for an intentional output change), from the repo root: + python tests/test_execution_details_golden.py +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any +from unittest.mock import MagicMock + +import pytest + +from tangle_cli.client import TangleApiClient +from tangle_cli.models import GetExecutionInfoResponse +from tangle_cli.pipeline_run_details import PipelineRunDetails + +FIXTURES = Path(__file__).parent / "fixtures" / "run_details" + + +def _leaf_ref(name: str, image: str) -> dict[str, Any]: + return { + "componentRef": { + "name": name, + "digest": f"digest-{name}", + "spec": { + "name": name, + "description": f"{name} does work", + "metadata": {"annotations": {"version": "1.0", "python_dependencies": "[big list]"}}, + "inputs": [{"name": "in", "type": "String"}], + "outputs": [{"name": "out", "type": "String"}], + "implementation": {"container": {"image": image, "command": ["run", name]}}, + }, + "text": f"raw yaml for {name}", + }, + "arguments": {"in": "hello"}, + } + + +def _execution( + exec_id: str, + name: str, + implementation: dict[str, Any], + *, + children: dict[str, str] | None = None, + inputs: dict[str, str] | None = None, + outputs: dict[str, str] | None = None, + state: str | None = None, +) -> dict[str, Any]: + result: dict[str, Any] = { + "id": exec_id, + "task_spec": { + "componentRef": { + "digest": f"digest-{exec_id}", + "spec": {"name": name, "inputs": [], "outputs": [], "implementation": implementation}, + "text": f"raw yaml for {name}", + }, + "arguments": {"arg": "value"}, + }, + "input_artifacts": {key: {"id": value} for key, value in (inputs or {}).items()}, + "output_artifacts": {key: {"id": value} for key, value in (outputs or {}).items()}, + } + if children: + result["child_task_execution_ids"] = children + if state: + result["state"] = state + return result + + +_FAN_GRAPH = { + "graph": { + "tasks": { + "Score A": _leaf_ref("Score", "gcr.io/score:v2"), + "Score B": _leaf_ref("Score", "gcr.io/score:v2"), + } + } +} + +#: root (graph) -> Prepare (leaf), Fan (graph) -> [Score A, Score B] (leaves), Report (leaf) +RESPONSES: dict[str, dict[str, Any]] = { + "root-exec": _execution( + "root-exec", + "Golden Pipeline", + { + "graph": { + "tasks": { + "Prepare": _leaf_ref("Prepare", "gcr.io/prep:v1"), + "Fan": { + "componentRef": { + "name": "Fan", + "digest": "digest-Fan", + "spec": {"name": "Fan", "implementation": _FAN_GRAPH}, + "text": "raw yaml for Fan", + }, + "arguments": {}, + }, + "Report": _leaf_ref("Report", "gcr.io/report:v1"), + } + } + }, + children={"Prepare": "prep-exec", "Fan": "fan-exec", "Report": "report-exec"}, + ), + "prep-exec": _execution( + "prep-exec", "Prepare", {"container": {"image": "gcr.io/prep:v1"}}, outputs={"out": "art-prep"}, + state="SUCCEEDED", + ), + "fan-exec": _execution( + "fan-exec", "Fan", _FAN_GRAPH, children={"Score A": "score-a-exec", "Score B": "score-b-exec"}, + inputs={"in": "art-prep"}, + ), + "score-a-exec": _execution( + "score-a-exec", "Score", {"container": {"image": "gcr.io/score:v2"}}, inputs={"in": "art-prep"}, + outputs={"out": "art-a"}, state="SUCCEEDED", + ), + "score-b-exec": _execution( + "score-b-exec", "Score", {"container": {"image": "gcr.io/score:v2"}}, inputs={"in": "art-prep"}, + outputs={"out": "art-b"}, state="FAILED", + ), + "report-exec": _execution( + "report-exec", "Report", {"container": {"image": "gcr.io/report:v1"}}, inputs={"a": "art-a", "b": "art-b"}, + ), +} + +VARIANTS: dict[str, dict[str, Any]] = { + "compact": {}, + "implementations": {"include_implementations": True}, + "state_and_annotations": {"include_execution_state": True, "include_annotations": True}, + "narrowed_to_subgraph": {"execution_id": "fan-exec"}, +} + + +def build_client() -> TangleApiClient: + client = TangleApiClient("https://api.test") + client.pipeline_runs_get = MagicMock( # type: ignore[method-assign] + return_value={ + "id": "run-1", + "root_execution_id": "root-exec", + "created_at": "2026-06-11T00:00:00Z", + "created_by": "tester@example.com", + "annotations": {"team": "search"}, + } + ) + + def details(execution_id: str) -> GetExecutionInfoResponse: + return GetExecutionInfoResponse.from_dict(json.loads(json.dumps(RESPONSES[execution_id]))) + + client.executions_details = MagicMock(side_effect=details) # type: ignore[method-assign] + client.pipeline_runs_annotations = MagicMock( # type: ignore[method-assign] + return_value={"team": "search", "owner": "tester"} + ) + stats = {"prep-exec": {"SUCCEEDED": 1}, "fan-exec": {"SUCCEEDED": 1, "FAILED": 1}} + client.executions_graph_execution_state = MagicMock( # type: ignore[method-assign] + return_value={"child_execution_status_stats": stats} + ) + return client + + +def render(variant: str) -> tuple[str, MagicMock]: + client = build_client() + output = PipelineRunDetails(client=client).get_run_details_output("run-1", **VARIANTS[variant]) + return json.dumps(output, indent=2, default=str) + "\n", client.executions_details # type: ignore[return-value] + + +@pytest.mark.parametrize("variant", sorted(VARIANTS)) +def test_run_details_output_is_byte_identical_to_golden(variant): + rendered, executions_details = render(variant) + + assert rendered == (FIXTURES / f"{variant}.json").read_text() + # Every execution in the (sub)tree is fetched exactly once. + expected = 3 if variant == "narrowed_to_subgraph" else len(RESPONSES) + assert executions_details.call_count == expected + assert len({call.args[0] for call in executions_details.call_args_list}) == expected + + +if __name__ == "__main__": # pragma: no cover - golden regeneration helper + FIXTURES.mkdir(parents=True, exist_ok=True) + for name in VARIANTS: + (FIXTURES / f"{name}.json").write_text(render(name)[0]) + print("wrote", FIXTURES / f"{name}.json") diff --git a/tests/test_execution_tree.py b/tests/test_execution_tree.py new file mode 100644 index 0000000..5c85441 --- /dev/null +++ b/tests/test_execution_tree.py @@ -0,0 +1,154 @@ +"""Tests for breadth-first execution-tree fetching (``tangle_cli.execution_tree``).""" + +from __future__ import annotations + +import threading +from typing import Any +from unittest.mock import MagicMock + +import pytest + +from tangle_cli import execution_tree as et +from tangle_cli.client import TangleApiClient +from tangle_cli.models import GetExecutionInfoResponse + +TREE: dict[str, dict[str, Any]] = { + "root": {"id": "root", "child_task_execution_ids": {"A": "a", "Sub": "sub", "Empty": None}}, + "a": {"id": "a"}, + "sub": {"id": "sub", "child_task_execution_ids": {"X": "x", "Y": "y"}}, + "x": {"id": "x"}, + "y": {"id": "y"}, +} +LEVELS = [ + ["/api/executions/root/details"], + ["/api/executions/a/details", "/api/executions/sub/details"], + ["/api/executions/x/details", "/api/executions/y/details"], +] + + +class Client: + def __init__(self) -> None: + self.calls: list[str] = [] + self.threads: set[int] = set() + self.prefetched: list[list[str]] = [] + self.persisted: list[list[str]] = [] + self._lock = threading.Lock() + + def executions_details(self, execution_id: str) -> dict[str, Any]: + with self._lock: + self.calls.append(execution_id) + self.threads.add(threading.get_ident()) + return dict(TREE[execution_id]) + + def prefetch(self, paths: list[str]) -> None: + self.prefetched.append(list(paths)) + + def persist(self, paths: list[str]) -> None: + self.persisted.append(list(paths)) + + +def test_breadth_first_with_names_parents_and_hooks(): + client = Client() + + tree = et.fetch_execution_tree(client, "root", max_workers=1) + + assert list(tree) == ["root", "a", "sub", "x", "y"] + assert [(n.name, n.parent) for n in tree.values()] == [ + (None, None), ("A", "root"), ("Sub", "root"), ("X", "sub"), ("Y", "sub"), + ] + assert client.prefetched == LEVELS + assert client.persisted == LEVELS + assert sorted(client.calls) == sorted(TREE) # each fetched once; the empty child id is skipped + + +def test_parallel_and_inline_fetching(monkeypatch): + parallel = Client() + et.fetch_execution_tree(parallel, "root", max_workers=4) + assert sorted(parallel.calls) == sorted(TREE) + + monkeypatch.setattr(et.sys, "platform", "emscripten") # no threads (e.g. Pyodide): never start any + inline = Client() + et.fetch_execution_tree(inline, "root", max_workers=8) + assert inline.threads == {threading.get_ident()} + + +def test_custom_fetch_and_errors_propagate(): + client = Client() + seen: list[str] = [] + + def fetch(execution_id: str) -> dict[str, Any]: + seen.append(execution_id) + if execution_id == "x": + raise RuntimeError("boom") + return dict(TREE[execution_id]) + + with pytest.raises(RuntimeError, match="boom"): + et.fetch_execution_tree(client, "root", fetch=fetch, max_workers=1) + assert client.calls == [] and "x" in seen + + +def test_child_execution_ids_reads_models_dicts_and_raw(): + class AttrOnly: + raw = None + child_task_execution_ids = {"B": "b", "A": "a"} + + class RawFirst: + raw = {"child_task_execution_ids": {"Z": "z"}} + child_task_execution_ids = {"ignored": "i"} + + assert et.child_execution_ids(AttrOnly()) == {"B": "b", "A": "a"} + assert et.child_execution_ids(RawFirst()) == {"Z": "z"} + assert et.child_execution_ids({"child_task_execution_ids": None}) == {} + assert et.child_execution_ids({}) == {} + + +def _model_client(**kwargs: Any) -> TangleApiClient: + client = TangleApiClient("https://api.test", **kwargs) + client.executions_details = MagicMock( # type: ignore[method-assign] + side_effect=lambda eid: GetExecutionInfoResponse.from_dict(dict(TREE[eid])) + ) + return client + + +def test_client_hooks_are_no_ops_by_default(): + client = TangleApiClient("https://api.test") + + assert client.prefetch(["/api/x"]) is None + assert client.persist(["/api/x"]) is None + + +def test_get_execution_details_fetches_breadth_first_and_calls_hooks(): + client = _model_client() + client.prefetch = MagicMock() # type: ignore[method-assign] + client.persist = MagicMock() # type: ignore[method-assign] + + details = client.get_execution_details("root") + + assert sorted(call.args[0] for call in client.executions_details.call_args_list) == sorted(TREE) + assert [call.args[0] for call in client.prefetch.call_args_list] == LEVELS + assert [call.args[0] for call in client.persist.call_args_list] == LEVELS + assert list(details.child_executions) == ["A", "Sub"] + assert list(details.child_executions["Sub"].child_executions) == ["X", "Y"] + + +def test_get_execution_details_respects_max_workers(monkeypatch): + client = _model_client() + client.execution_tree_max_workers = 1 + seen: list[int] = [] + monkeypatch.setattr(et, "parallel_map", lambda fn, items, workers: seen.append(workers) or [fn(i) for i in items]) + + client.get_execution_details("root") + + assert seen and set(seen) == {1} + + +def test_enrich_without_prefetched_tree_still_fetches_children(): + """Subclasses calling ``_enrich_execution_tree(execution)`` keep the per-node fetch.""" + + client = _model_client() + root = GetExecutionInfoResponse.from_dict(dict(TREE["root"])) + + client._enrich_execution_tree(root) + + assert sorted(call.args[0] for call in client.executions_details.call_args_list) == ["a", "sub", "x", "y"] + assert list(root.child_executions) == ["A", "Sub"] diff --git a/tests/test_packaging.py b/tests/test_packaging.py index 08e316c..b64eea2 100644 --- a/tests/test_packaging.py +++ b/tests/test_packaging.py @@ -183,7 +183,7 @@ def test_tangle_cli_wheel_supports_expert_no_deps_import_path_without_tangle_api requires_dist = [line for line in metadata.splitlines() if line.startswith("Requires-Dist: ")] assert not any(name.startswith("tangle_api/") for name in names) assert "tangle_cli/openapi/openapi.json" not in names - assert "Version: 0.1.33" in metadata + assert "Version: 0.1.34" in metadata assert "Requires-Dist: tangle-api==0.1.1" in requires_dist assert not any("extra == 'native'" in line for line in requires_dist) assert "Provides-Extra: native" in metadata diff --git a/tests/test_rate_limit.py b/tests/test_rate_limit.py new file mode 100644 index 0000000..f87a7a0 --- /dev/null +++ b/tests/test_rate_limit.py @@ -0,0 +1,392 @@ +"""Tests for client-side request pacing (``tangle_cli.rate_limit`` + client integration).""" + +from __future__ import annotations + +from typing import Any + +import pytest +import requests + +from tangle_cli import client as client_module +from tangle_cli import rate_limit as rl +from tangle_cli.client import TangleApiClient + + +class FakeClock: + """Deterministic clock: ``sleep`` advances time and records the wait.""" + + def __init__(self) -> None: + self.now = 100.0 + self.sleeps: list[float] = [] + + def clock(self) -> float: + return self.now + + def sleep(self, seconds: float) -> None: + self.sleeps.append(round(seconds, 6)) + self.now += seconds + + +def _limiter(rps: float | None, clock: FakeClock, *, thread: list[int] | None = None, rand: float = 0.5): + current = thread if thread is not None else [1] + return rl.RequestRateLimiter( + rps, clock=clock.clock, sleep=clock.sleep, rand=lambda: rand, thread_id=lambda: current[0] + ) + + +def test_requests_are_spaced_at_the_configured_rate(): + clock = FakeClock() + limiter = _limiter(8, clock) + + for _ in range(5): + limiter.acquire() + + assert clock.sleeps == [0.125, 0.125, 0.125, 0.125] # first slot is immediate + assert limiter.max_per_second == 8 + + +def test_idle_time_is_not_banked_into_a_burst(): + clock = FakeClock() + limiter = _limiter(4, clock) + + limiter.acquire() + clock.now += 10 # long idle + limiter.acquire() + limiter.acquire() + + assert clock.sleeps == [0.25] + + +@pytest.mark.parametrize("rps", [None, 0, -1]) +def test_disabled_pacing_never_waits(rps): + clock = FakeClock() + limiter = _limiter(rps, clock) + + for _ in range(10): + assert limiter.acquire() == 0.0 + + assert clock.sleeps == [] and limiter.max_per_second is None + + +def test_pause_holds_other_threads_with_jitter_but_not_the_pausing_thread(): + clock = FakeClock() + thread = [1] + limiter = _limiter(None, clock, thread=thread, rand=0.5) + + limiter.pause(5.0) # thread 1 got a 429 and sleeps through its own backoff + assert limiter.acquire() == 0.0 # thread 1 itself is exempt + + thread[0] = 2 + waited = limiter.acquire() # another worker waits for the pause + jitter + + assert waited == pytest.approx(5.0 + 0.5 * min(rl.PAUSE_JITTER_MAX_SECONDS, rl.PAUSE_JITTER_FRACTION * 5.0)) + thread[0] = 3 + assert limiter.acquire() == 0.0 # the pause is over + + +def test_pause_combines_with_pacing_and_only_extends(): + clock = FakeClock() + thread = [1] + limiter = _limiter(2, clock, thread=thread, rand=0.0) + + limiter.pause(3.0) + limiter.pause(1.0) # a shorter pause never shortens the active one + thread[0] = 2 + first = limiter.acquire() + second = limiter.acquire() + + assert first == pytest.approx(3.0) + assert second == pytest.approx(0.5) # back to the 2 req/s spacing after the pause + limiter.pause(0) # no-op + limiter.pause(-1) + + +def test_a_pause_that_starts_while_a_caller_sleeps_still_holds_it(): + clock = FakeClock() + thread = [2] + limiter = rl.RequestRateLimiter( + 8, clock=clock.clock, rand=lambda: 0.5, thread_id=lambda: thread[0], sleep=None + ) + paused = [] + + def sleep(seconds: float) -> None: + clock.sleep(seconds) + if not paused: # another worker gets a 429 (Retry-After: 10) while this one sleeps + paused.append(True) + thread[0] = 1 + limiter.pause(10.0) + thread[0] = 2 + + limiter._sleep = sleep + limiter.acquire() # slot at t=100 + waited = limiter.acquire() # sleeps 0.125 for its slot, then must honour the new pause + + # 0.125 for the slot, then 10s pause + jitter (min(2, 0.2*10) * 0.5 = 1.0). + assert waited == pytest.approx(0.125 + 11.0) + assert clock.now == pytest.approx(100.0 + 11.125) + + +def test_a_lower_live_rate_applies_to_the_very_next_request(): + clock = FakeClock() + limiter = _limiter(8, clock) + + limiter.acquire() + limiter.set_rate(4) # caller reacts to pushback by slowing down + limiter.acquire() + limiter.set_rate(None) + limiter.acquire() + + assert clock.sleeps == [0.25] # new 4 req/s spacing, then no pacing at all + + +def test_a_slot_beyond_the_deadline_is_refused_without_sleeping(): + clock = FakeClock() + thread = [1] + limiter = _limiter(None, clock, thread=thread, rand=0.0) + + limiter.pause(60.0) # another worker got a 429 with Retry-After: 60 + thread[0] = 2 + + assert limiter.acquire(max_wait=30.0) is None + assert clock.sleeps == [] + assert limiter.acquire(max_wait=90.0) == pytest.approx(60.0) + + +def test_set_rate_keeps_an_active_pause(): + clock = FakeClock() + thread = [1] + limiter = _limiter(8, clock, thread=thread, rand=0.0) + + limiter.pause(4.0) + limiter.set_rate(2) + thread[0] = 2 + + assert limiter.acquire() == pytest.approx(4.0) + assert limiter.max_per_second == 2 + + +class _Session: + def __init__(self, responses: list[requests.Response]) -> None: + self.responses = responses + self.calls = 0 + + def request(self, method: str, url: str, **kwargs: Any) -> requests.Response: + self.calls += 1 + return self.responses.pop(0) + + +def _response(status: int, headers: dict[str, str] | None = None) -> requests.Response: + response = requests.Response() + response.status_code = status + response._content = b"{}" + response.headers.update(headers or {}) + return response + + +def _client(responses: list[requests.Response], **kwargs: Any) -> TangleApiClient: + return TangleApiClient("https://api.test", session=_Session(responses), **kwargs) + + +def test_client_default_and_override(): + assert TangleApiClient.DEFAULT_MAX_REQUESTS_PER_SECOND == 8.0 # CPython default + assert _client([])._request_rate_limiter().max_per_second == 8.0 + assert _client([], max_requests_per_second=None)._request_rate_limiter().max_per_second is None + + client = _client([], max_requests_per_second=2) + first = client._request_rate_limiter() + assert first.max_per_second == 2 and client._request_rate_limiter() is first # one shared limiter + first.pause(30.0) + client.max_requests_per_second = 5 + updated = client._request_rate_limiter() + assert updated is first and updated.max_per_second == 5 # updated in place + assert updated._paused_until > 0 # the active pause survives the change + + +def test_every_physical_send_acquires_a_slot(monkeypatch): + monkeypatch.setattr("tangle_cli.client.time.sleep", lambda _seconds: None) + client = _client([_response(503), _response(200)], max_requests_per_second=None) + acquired: list[str] = [] + limiter = client._request_rate_limiter() + monkeypatch.setattr(limiter, "acquire", lambda **_: acquired.append("slot") or 0.0) + + response = client._make_request("GET", "/api/x") + + assert response.status_code == 200 + assert acquired == ["slot", "slot"] # the retry is paced too + + +def test_limiter_wait_happens_before_the_send_budget_deadline_check(monkeypatch): + """A send delayed by pacing past the retry deadline is refused, not sent late.""" + + client = _client([_response(200)], max_requests_per_second=None) + events: list[str] = [] + limiter = client._request_rate_limiter() + monkeypatch.setattr(limiter, "acquire", lambda **_: events.append("acquire") or 0.0) + real_budget = client_module._RetryBudget + + class RecordingBudget(real_budget): # type: ignore[misc, valid-type] + def try_consume_send(self) -> bool: + events.append("budget") + return super().try_consume_send() + + monkeypatch.setattr(client_module, "_RetryBudget", RecordingBudget) + + assert client._make_request("GET", "/api/x").status_code == 200 + assert events == ["acquire", "budget"] + + +def _slow_limiter(monkeypatch, client: TangleApiClient, seconds: float) -> list[float]: + """Fake clock + a limiter that holds every slot for *seconds*.""" + + monkeypatch.setattr(client, "_MAX_RETRY_ELAPSED_SECONDS", 5.0) + monkeypatch.setattr("tangle_cli.client.time.sleep", lambda _seconds: None) + now = [1000.0] + monkeypatch.setattr(client_module.time, "monotonic", lambda: now[0]) + + def slow_acquire(max_wait: float | None = None) -> float | None: + now[0] += seconds + return seconds + + monkeypatch.setattr(client._request_rate_limiter(), "acquire", slow_acquire) + return now + + +def test_queueing_for_the_first_slot_does_not_spend_the_retry_deadline(monkeypatch): + """A worker that waits behind others for its first send is not retrying.""" + + session = _Session([_response(200)]) + client = TangleApiClient("https://api.test", session=session, max_requests_per_second=None) + _slow_limiter(monkeypatch, client, 10.0) # first slot: 10s, past the 5s retry deadline + + assert client._make_request("GET", "/api/x").status_code == 200 + assert session.calls == 1 + + +def test_a_retry_whose_pacing_wait_outlasts_the_deadline_is_not_sent(monkeypatch): + session = _Session([_response(503), _response(200)]) + client = TangleApiClient("https://api.test", session=session, max_requests_per_second=None) + _slow_limiter(monkeypatch, client, 10.0) + + response = client._make_request("GET", "/api/x") + + assert response.status_code == 503 # the retry was refused at the deadline, not sent late + assert session.calls == 1 + + +def test_auth_refresh_and_header_snapshots_are_serialized(monkeypatch): + events: list[str] = [] + + class RecordingLock: + def __enter__(self): + events.append("enter") + + def __exit__(self, *exc): + events.append("exit") + + client = _client([_response(200)], max_requests_per_second=None) + client.__dict__["_state_lock"] = RecordingLock() + monkeypatch.setattr(client, "_refresh_auth", lambda: events.append("refresh")) + real_headers = client._headers + monkeypatch.setattr(client, "_headers", lambda extra=None: events.append("headers") or real_headers(extra)) + + client._make_request("GET", "/api/x") + + assert events == ["enter", "refresh", "exit", "enter", "headers", "exit"] + + +def test_first_slot_is_taken_after_the_auth_refresh(monkeypatch): + """Workers released together from a slow, serialized refresh are still paced.""" + + events: list[str] = [] + client = _client([_response(200)], max_requests_per_second=None) + monkeypatch.setattr(client, "_refresh_auth", lambda: events.append("refresh")) + monkeypatch.setattr(client._request_rate_limiter(), "acquire", lambda **_: events.append("acquire") or 0.0) + real_request = client.session.request + monkeypatch.setattr(client.session, "request", lambda *a, **k: events.append("send") or real_request(*a, **k)) + + client._make_request("GET", "/api/x") + + assert events == ["refresh", "acquire", "send"] + + +def test_client_state_lock_is_reentrant_and_shared(): + client = _client([]) + lock = client._client_state_lock() + + assert client._client_state_lock() is lock + with lock, lock: # a refresh that issues requests through the client re-enters it + pass + + +def test_a_retry_slot_beyond_the_deadline_returns_the_last_answer_without_waiting(monkeypatch): + session = _Session([_response(503), _response(200)]) + client = TangleApiClient("https://api.test", session=session, max_requests_per_second=None) + monkeypatch.setattr("tangle_cli.client.time.sleep", lambda _seconds: None) + limiter = client._request_rate_limiter() + seen: list[float | None] = [] + + def acquire(max_wait: float | None = None) -> float | None: + seen.append(max_wait) + return None if max_wait is not None else 0.0 # the retry's slot is past the deadline + + monkeypatch.setattr(limiter, "acquire", acquire) + + response = client._make_request("GET", "/api/x") + + assert response.status_code == 503 and session.calls == 1 + assert seen[0] is None and seen[1] is not None # first slot unbounded, retry slot bounded + + +def test_streamed_open_honours_503_retry_after_and_pauses(monkeypatch): + sleeps: list[float] = [] + monkeypatch.setattr("tangle_cli.client.time.sleep", sleeps.append) + + class StreamSession(_Session): + def request(self, method: str, url: str, **kwargs: Any) -> requests.Response: + response = super().request(method, url, **kwargs) + response._content_consumed = True # buffered fake: close() has no raw stream to release + return response + + busy = _response(503, {"Retry-After": "12"}) + ok = _response(200) + client = TangleApiClient("https://api.test", session=StreamSession([busy, ok]), max_requests_per_second=None) + pauses: list[float] = [] + monkeypatch.setattr(client._request_rate_limiter(), "pause", pauses.append) + monkeypatch.setattr(client, "_make_stream_body_idle_unbounded", lambda _response: None) + + assert client.stream_execution_container_log("exec-1").status_code == 200 + assert sleeps == [12.0] and pauses == [12.0] + + +def test_429_pauses_the_shared_limiter_for_the_retry_after(monkeypatch): + sleeps: list[float] = [] + monkeypatch.setattr("tangle_cli.client.time.sleep", sleeps.append) + client = _client([_response(429, {"Retry-After": "7"}), _response(200)], max_requests_per_second=None) + pauses: list[float] = [] + monkeypatch.setattr(client._request_rate_limiter(), "pause", pauses.append) + + assert client._make_request("GET", "/api/x").status_code == 200 + assert pauses == [7.0] + assert sleeps == [7.0] # the retrying request's own wait is unchanged + + +def test_503_honours_retry_after_and_pauses(monkeypatch): + sleeps: list[float] = [] + monkeypatch.setattr("tangle_cli.client.time.sleep", sleeps.append) + client = _client([_response(503, {"Retry-After": "4"}), _response(200)], max_requests_per_second=None) + pauses: list[float] = [] + monkeypatch.setattr(client._request_rate_limiter(), "pause", pauses.append) + + assert client._make_request("GET", "/api/x").status_code == 200 + assert pauses == [4.0] and sleeps == [4.0] + + +def test_other_5xx_keep_their_backoff_without_pausing(monkeypatch): + sleeps: list[float] = [] + monkeypatch.setattr("tangle_cli.client.time.sleep", sleeps.append) + client = _client([_response(502, {"Retry-After": "9"}), _response(200)], max_requests_per_second=None) + pauses: list[float] = [] + monkeypatch.setattr(client._request_rate_limiter(), "pause", pauses.append) + + assert client._make_request("GET", "/api/x").status_code == 200 + assert pauses == [] and sleeps == [1.0] diff --git a/uv.lock b/uv.lock index 62c2baa..61c05dc 100644 --- a/uv.lock +++ b/uv.lock @@ -2083,7 +2083,7 @@ requires-dist = [{ name = "pydantic", specifier = ">=2.0" }] [[package]] name = "tangle-cli" -version = "0.1.33" +version = "0.1.34" source = { editable = "." } dependencies = [ { name = "cloud-pipelines" },