diff --git a/CHANGELOG.md b/CHANGELOG.md index d835e16..46477c5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,8 +14,53 @@ quickstart, guides, full API reference, and `llms.txt`; deployed to GitHub Pages on pushes to `main`. +### Added + +- Concurrency guide covering what is safe to share, how stale models + work, and the lost-update windows the Codesphere API provides no way + to close (it has no ETags, `If-Match`, or idempotency keys). +- `Workspace.refresh()` and `Domain.refresh()` re-read an entity from the + server in place. + +### Fixed + +- `wait_for_stage()` pins the pipeline run it is waiting on by its start + time and raises `ConflictError` if someone else restarts the stage + meanwhile, instead of silently reporting the outcome of a run the + caller never started. +- `wait_for_stage()` and `wait_until_running()` now measure their + `timeout` against the wall clock. Previously only the sleeps counted, + so slow status requests could overrun the deadline substantially. +- Reusing a `LogStream` raises `ClientStateError` instead of stranding + the first stream context or surfacing a raw `httpx.StreamConsumed`. +- Sharing one SDK across tasks or threads is now safe. `open()`/`close()` + are reference counted, so nested and concurrent scopes no longer tear + the transport down for each other, and two threads racing `open()` in + the sync client can no longer each build (and orphan) a connection + pool. Closing while a request is in flight now raises the new + `ClientStateError` instead of leaking httpx's bare `RuntimeError`. +- `sdk.flags.invalidate()` is no longer discarded when a flags fetch is + already in flight; the stale snapshot used to be reinstated silently. + The `legacy_platform` marker in feature-flag errors can also no longer + come from a different fetch than the snapshot it is reported with. +- A `404` on a retried `DELETE` is treated as success. Previously a + teardown that had actually succeeded reported `NotFoundError` when the + first attempt's response was lost behind a gateway error. A `404` on + the first attempt still raises. + ### Changed +- **Breaking:** writes no longer copy values back into the local model. + `Workspace.update()`, `Domain.update()`, + `Domain.update_workspace_connections()` and `Domain.verify_status()` + now mark the instance **stale**: reading a field, `to_dict()`, + `to_json()` or `to_yaml()` raises the new `StaleModelError` until you + call `await refresh()`. The platform orders writes by arrival while the + client sees them by response, so the old write-back could leave a model + reporting a value the platform did not hold. Identity fields (`id`, or + `name`/`team_id` for domains) stay readable. `Domain` methods still + return the server's response, which is authoritative — prefer it over + re-reading `self`. - Retries on transient failures are now enabled by default (`max_retries=2`). Idempotent methods (`GET`, `HEAD`, `PUT`, `DELETE`) are retried on `429`/`502`/`503`/`504` and connect/timeout errors. @@ -32,6 +77,8 @@ - The deprecated module path `codesphere.resources.workspace.envVars` (use `codesphere.resources.workspace.env_vars`). +- `codesphere.utils.update_model_fields`, the helper behind the removed + write-back behavior. It had no remaining callers. ## v1.0.0 (2026-02-21) diff --git a/docs/guides/concurrency.md b/docs/guides/concurrency.md new file mode 100644 index 0000000..e80d567 --- /dev/null +++ b/docs/guides/concurrency.md @@ -0,0 +1,162 @@ +# Concurrency + +The async client is built to be shared. This guide covers what the SDK +guarantees when several tasks use it at once, and — just as important — +what it cannot guarantee, because the Codesphere API offers no way to. + +## Share one client + +Create one `CodesphereSDK` and use it from as many tasks as you like. +Sharing is preferred: each instance owns a connection pool, so creating +one per task throws away connection reuse. + +```python +import asyncio +from codesphere import CodesphereSDK + +async def main(): + async with CodesphereSDK() as sdk: + teams = await sdk.teams.list() + # One client, many concurrent calls. + workspaces = await asyncio.gather( + *(sdk.workspaces.list(team.id) for team in teams) + ) +``` + +Scopes are reference counted, so nested and concurrent `async with` +blocks are safe — the transport closes when the last one exits, not the +first: + +```python +async def worker(sdk): + async with sdk: # each worker holds its own scope + await sdk.teams.list() + +async with CodesphereSDK() as sdk: + await asyncio.gather(worker(sdk), worker(sdk)) + await sdk.teams.list() # still open +``` + +Using the client after it is fully closed raises +[`ClientStateError`][codesphere.ClientStateError], which is both a +`CodesphereError` and a `RuntimeError`. + +The synchronous client offers the same guarantees across threads. + +## Models go stale after a write + +The platform applies writes in the order they arrive; your program sees +them in the order responses come back. Those orders can differ, so the +SDK cannot know an entity's state after a write it did not read back. +Rather than report a value the platform may not hold, a write +**invalidates** the instance: + +```python +workspace = await sdk.workspaces.get(72678) + +await workspace.update(WorkspaceUpdate(name="renamed")) + +workspace.id # fine: a write cannot change an entity's identity +workspace.name # raises StaleModelError +``` + +Call `refresh()` to re-read the server's actual state: + +```python +await workspace.refresh() +workspace.name # server-authoritative +``` + +This applies to `Workspace.update()`, `Domain.update()`, +`Domain.update_workspace_connections()` and `Domain.verify_status()`. +`to_dict()`, `to_json()` and `to_yaml()` raise on a stale model too. +The `Domain` methods return the server's response — prefer that returned +object over re-reading the instance you called them on. + +## What the SDK cannot protect you from + +The workspace you are reading can be changed at the same moment by the +web IDE, a CI pipeline, a teammate, or another process of your own. The +Codesphere API has **no ETags, no `If-Match`, and no idempotency keys**, +so there is no way for the SDK to detect or reject a conflicting write. +These are real limitations, not oversights, and the SDK does not pretend +otherwise. + +### Read-modify-write loses concurrent changes + +Any read, edit, write cycle has a window in which someone else's change +is silently overwritten: + +```python +# NOT safe against concurrent writers +current = await workspace.env_vars.get() +await workspace.env_vars.set([*current, EnvVar(name="NEW", value="1")]) +``` + +`env_vars.set()` is a full replacement (`PUT`), so a variable added by +someone else between the two calls is erased. The same applies to +`landscape.get_profile()` → edit → `save_profile()`, which writes through +a shell redirect and is not an atomic file replacement. + +If you must do this, narrow the window and verify afterwards by reading +back. There is no way to make it atomic. + +### Retried writes can execute twice + +`PUT` and `DELETE` are idempotent as HTTP methods but not always in +effect. If the platform receives a request and only the response is lost +(a gateway `503`, a dropped connection), the retry runs the operation +again — a landscape teardown can happen twice. Without idempotency keys +the SDK cannot deduplicate this. + +The most common symptom is handled: a `404` on a **retried** `DELETE` is +treated as success, since the resource being gone is what you asked for. +A `404` on the first attempt still raises `NotFoundError`. + +If duplicate execution is unacceptable, disable retries for those calls: + +```python +sdk = CodesphereSDK(retry=RetryConfig(max_retries=0)) +``` + +### Waiting on a pipeline someone else restarted + +`wait_for_stage()` pins the run it is watching by its start time. If +someone redeploys mid-wait, it raises +[`ConflictError`][codesphere.ConflictError] instead of reporting the +outcome of a run you never started: + +```python +try: + await workspace.landscape.wait_for_stage("run", timeout=600) +except ConflictError: + # Someone else redeployed. Decide whether to wait on the new run. + ... +``` + +Timeouts on `wait_for_stage()` and `wait_until_running()` are wall clock: +time spent inside the status requests counts against your budget. + +## Log streams are single-use + +`logs.open()` returns a stream backed by one SSE response body, which can +only be read once. Opening or iterating the same stream twice raises +`ClientStateError`. Call `logs.open()` again for a second stream — +concurrent streams over one client are fine: + +```python +async def tail(target): + async for entry in workspace.logs.stream(target): + print(entry.message) + +await asyncio.gather( + tail(ServerTarget(step=1, server="web")), + tail(ServerTarget(step=1, server="api")), +) +``` + +## Feature flags + +The flags snapshot is fetched once per client and cached. Concurrent +gated calls collapse into a single request, and `invalidate()` is honored +even against a fetch that is already in flight. diff --git a/docs/llms.txt b/docs/llms.txt index 059dde5..7eaa3d4 100644 --- a/docs/llms.txt +++ b/docs/llms.txt @@ -26,6 +26,7 @@ keyword-only `timeout=` override. Errors derive from - [Retries & timeouts](https://datata1.github.io/codesphere-python/guides/retries/): default retry behavior, RetryConfig, per-call timeout - [Feature flags](https://datata1.github.io/codesphere-python/guides/feature-flags/): sdk.flags, operation gating errors - [Streaming logs](https://datata1.github.io/codesphere-python/guides/streaming-logs/): SSE log streaming, targets, deadlines +- [Concurrency](https://datata1.github.io/codesphere-python/guides/concurrency/): sharing one client, stale models after writes, lost-update windows the API cannot close - [Sync vs Async](https://datata1.github.io/codesphere-python/guides/sync-vs-async/): choosing a flavor, shared models, caveats - [API Reference](https://datata1.github.io/codesphere-python/reference/client/): full typed API surface - [Changelog](https://datata1.github.io/codesphere-python/changelog/): release history diff --git a/docs/reference/config.md b/docs/reference/config.md index 799b7e8..cc95c6a 100644 --- a/docs/reference/config.md +++ b/docs/reference/config.md @@ -20,6 +20,8 @@ - RateLimitError - NetworkError - TimeoutError + - ClientStateError + - StaleModelError - FeatureFlagError - FeatureNotAvailableError - FeatureNotEnabledError diff --git a/mkdocs.yml b/mkdocs.yml index 76d0dfa..718b845 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -54,6 +54,7 @@ nav: - Retries & timeouts: guides/retries.md - Feature flags: guides/feature-flags.md - Streaming logs: guides/streaming-logs.md + - Concurrency: guides/concurrency.md - Sync vs Async: guides/sync-vs-async.md - API Reference: - Client: reference/client.md diff --git a/scripts/gen_sync.py b/scripts/gen_sync.py index c40d574..5f8f803 100644 --- a/scripts/gen_sync.py +++ b/scripts/gen_sync.py @@ -50,7 +50,10 @@ def restore_all_blocks() -> int: def main() -> None: - run("uv", "run", "unasyncd") + # unasyncd exits non-zero whenever it rewrites a file, the way a + # formatter signals "changed". That is the normal case here, so it + # must not abort the post-processing below. + run("uv", "run", "unasyncd", fatal=False) restored = restore_all_blocks() print(f"Restored __all__ in {restored} generated files") # Remaining findings are silenced via per-file-ignores in ruff.toml; diff --git a/src/codesphere/__init__.py b/src/codesphere/__init__.py index 5b3aa2a..2b7d456 100644 --- a/src/codesphere/__init__.py +++ b/src/codesphere/__init__.py @@ -33,6 +33,7 @@ APIError, AuthenticationError, AuthorizationError, + ClientStateError, CodesphereError, ConflictError, FeatureFlagError, @@ -41,6 +42,7 @@ NetworkError, NotFoundError, RateLimitError, + StaleModelError, TimeoutError, ValidationError, ) @@ -73,6 +75,7 @@ "AuthorizationError", "CategoryFlags", "Characteristic", + "ClientStateError", "CodesphereError", "CodesphereSDK", "ConflictError", @@ -93,6 +96,7 @@ "NotFoundError", "RateLimitError", "RetryConfig", + "StaleModelError", "SyncCodesphereSDK", "Team", "TeamBase", diff --git a/src/codesphere/_async/core/base.py b/src/codesphere/_async/core/base.py index 36f3dd3..5eb0f1c 100644 --- a/src/codesphere/_async/core/base.py +++ b/src/codesphere/_async/core/base.py @@ -1,5 +1,5 @@ from collections.abc import Mapping -from typing import Any, TypeVar +from typing import Any, ClassVar, TypeVar import httpx from pydantic import PrivateAttr @@ -7,6 +7,7 @@ from codesphere.core.models import CamelModel from codesphere.core.models import ResourceList as ResourceList from codesphere.core.operations import APIOperation +from codesphere.exceptions import StaleModelError from codesphere.feature_flags import FlagRequirement from ..http_client import APIHttpClient @@ -66,9 +67,78 @@ class BoundModel(CamelModel): Instances returned by the SDK get their client attached automatically, which lets entity methods (e.g. ``workspace.delete()``) make further API calls. + + A write that the SDK cannot verify marks the instance **stale**: its + field values are dropped and reads raise + :class:`~codesphere.StaleModelError` until ``refresh()`` re-reads the + entity. Identity fields survive, so the instance can still be logged + and re-fetched. See :meth:`_mark_stale`. """ + #: Fields that stay readable on a stale instance. An entity's identity + #: cannot be changed by a write, so it is always safe to report. + _identity_fields: ClassVar[tuple[str, ...]] = ("id",) + _http_client: APIHttpClient | None = PrivateAttr(default=None) + _stale: bool = PrivateAttr(default=False) + + def __getattr__(self, item: str) -> Any: + # Pydantic keeps field values in __dict__, so this only runs once a + # lookup has already failed: zero cost until a model goes stale. + if self._is_stale() and item in type(self).__pydantic_fields__: + raise StaleModelError(type(self).__name__, item) + # Delegate to pydantic, which resolves private attributes here. + # It defines __getattr__ only at runtime, so it is fetched + # dynamically rather than called through super() directly. + parent = getattr(super(), "__getattr__", None) + if parent is None: # pragma: no cover - pydantic always defines it + raise AttributeError(item) + return parent(item) + + def _is_stale(self) -> bool: + private = object.__getattribute__(self, "__pydantic_private__") + return bool(private and private.get("_stale")) + + def _mark_stale(self) -> None: + """Drop local field values the SDK can no longer vouch for. + + Called after a write whose resulting server state is unknown. + Clearing ``__dict__`` (rather than setting a flag beside intact + values) is what routes later reads through ``__getattr__``. + """ + identity = { + name: value + for name, value in self.__dict__.items() + if name in type(self)._identity_fields + } + # Also drops cached_property managers; they rebuild after refresh. + self.__dict__.clear() + self.__dict__.update(identity) + self._stale = True + + def _adopt(self, fresh: "BoundModel") -> None: + """Repopulate from a server-authoritative re-read.""" + self.__dict__.clear() + self.__dict__.update(fresh.__dict__) + self._stale = False + + def model_dump(self, *args: Any, **kwargs: Any) -> dict[str, Any]: + # Pydantic serializes straight from __dict__, so without this a + # stale model would quietly dump only its identity fields. + if self._is_stale(): + raise StaleModelError(type(self).__name__) + return super().model_dump(*args, **kwargs) + + def model_dump_json(self, *args: Any, **kwargs: Any) -> str: + if self._is_stale(): + raise StaleModelError(type(self).__name__) + return super().model_dump_json(*args, **kwargs) + + def __repr_args__(self) -> Any: + # Keep repr() working for debugging, but say why it looks empty. + if self._is_stale(): + return [*super().__repr_args__(), ("stale", True)] + return super().__repr_args__() def _client(self) -> APIHttpClient: if self._http_client is None or not hasattr(self._http_client, "request"): diff --git a/src/codesphere/_async/flags_store.py b/src/codesphere/_async/flags_store.py index 1b0ad1b..a148bc0 100644 --- a/src/codesphere/_async/flags_store.py +++ b/src/codesphere/_async/flags_store.py @@ -27,6 +27,11 @@ class FlagsStore: client's lifetime. A 404 — the instance predates feature flags — is cached as an empty snapshot: reads report nothing available and gated calls fail closed. Transient errors propagate uncached. + + Safe to share across tasks: the snapshot and the ``legacy_platform`` + flag it was fetched with are always read as one pair, and a fetch + that was in flight when :meth:`invalidate` ran does not repopulate + the cache it was told to drop. """ def __init__(self, client: "APIHttpClient") -> None: @@ -34,25 +39,43 @@ def __init__(self, client: "APIHttpClient") -> None: self._snapshot: FlagsSnapshot | None = None self._legacy_platform = False self._lock = Lock() + # Bumped by invalidate(). A fetch stamps the generation it started + # under and refuses to install its result if that has moved on. + self._generation = 0 async def get(self, *, refresh: bool = False) -> FlagsSnapshot: + snapshot, _ = await self._get_with_legacy(refresh=refresh) + return snapshot + + async def _get_with_legacy( + self, *, refresh: bool = False + ) -> tuple[FlagsSnapshot, bool]: + """The snapshot and the legacy-platform flag it was fetched with. + + Returned together from one lock acquisition so callers cannot pair + a snapshot with a legacy flag from a different fetch. + """ async with self._lock: - if self._snapshot is None or refresh: - self._snapshot = await self._fetch() - return self._snapshot + if self._snapshot is not None and not refresh: + return self._snapshot, self._legacy_platform - async def _fetch(self) -> FlagsSnapshot: + generation = self._generation + snapshot, legacy = await self._fetch() + if generation == self._generation: + self._snapshot = snapshot + self._legacy_platform = legacy + return snapshot, legacy + + async def _fetch(self) -> tuple[FlagsSnapshot, bool]: try: response = await self._client.request("GET", FLAGS_ENDPOINT) except NotFoundError: - self._legacy_platform = True - return FlagsSnapshot.empty() - self._legacy_platform = False - return FlagsSnapshot.model_validate(response.json()) + return FlagsSnapshot.empty(), True + return FlagsSnapshot.model_validate(response.json()), False async def require(self, requirement: FlagRequirement) -> None: """Raise unless the required flag is enabled on this instance.""" - snapshot = await self.get() + snapshot, legacy_platform = await self._get_with_legacy() flag, category = requirement.flag, requirement.category category_str = str(category) if category is not None else None @@ -60,7 +83,7 @@ async def require(self, requirement: FlagRequirement) -> None: raise FeatureNotAvailableError( flag=flag, category=category_str, - legacy_platform=self._legacy_platform, + legacy_platform=legacy_platform, ) if not snapshot.is_enabled(flag, category): raise FeatureNotEnabledError(flag=flag, category=category_str) @@ -70,5 +93,11 @@ def cached(self) -> FlagsSnapshot | None: return self._snapshot def invalidate(self) -> None: - """Drop the cached snapshot; the next need re-fetches.""" + """Drop the cached snapshot; the next need re-fetches. + + Stays synchronous on purpose: acquiring the lock here would block + the caller for as long as an in-flight flags request takes. The + generation bump is what makes that safe. + """ + self._generation += 1 self._snapshot = None diff --git a/src/codesphere/_async/http_client.py b/src/codesphere/_async/http_client.py index 554118b..7ff8394 100644 --- a/src/codesphere/_async/http_client.py +++ b/src/codesphere/_async/http_client.py @@ -11,8 +11,14 @@ from pydantic import SecretStr from codesphere.config import RetryConfig -from codesphere.exceptions import NetworkError, TimeoutError, raise_for_status - +from codesphere.exceptions import ( + ClientStateError, + NetworkError, + TimeoutError, + raise_for_status, +) + +from ._compat import Lock from .flags_store import FlagsStore log = logging.getLogger(__name__) @@ -31,12 +37,25 @@ def _parse_retry_after(value: str | None) -> float | None: return max(0.0, (retry_at - datetime.now(UTC)).total_seconds()) +def _is_delete_of_gone_resource(method: str, status_code: int, attempt: int) -> bool: + """Whether a 404 means "the DELETE already succeeded" rather than a miss. + + Only ever true after at least one retry: a 404 on the very first + attempt is a genuine "no such resource" and must still raise. + """ + return attempt > 0 and status_code == 404 and method.upper() == "DELETE" + + class APIHttpClient: """Thin transport around httpx. Receives its full configuration from the caller (the SDK resolves arguments, environment variables, and defaults) and owns nothing but the connection lifecycle, error mapping, and opt-in retries. + + Safe to share across tasks. ``open()``/``close()`` are reference + counted, so nested and concurrent scopes each hold the transport open + and only the last one to exit tears it down. """ def __init__( @@ -51,6 +70,11 @@ def __init__( self._base_url = base_url self._client: httpx.AsyncClient | None = None self._retry = retry if retry is not None else RetryConfig() + # Guards the check-then-act in open()/close(). Without it, two + # threads in the sync twin can each build a transport and orphan + # one of them. + self._lifecycle_lock = Lock() + self._scopes = 0 # Per-connection feature-flags cache; the transport never reads it # itself (see feature_flags.FlagsStore). self.flags_store = FlagsStore(self) @@ -64,10 +88,7 @@ def __init__( def _get_client(self) -> httpx.AsyncClient: if not self._client: - raise RuntimeError( - "Client is not open. Use the client as a context manager " - "or call open() before making requests." - ) + raise ClientStateError() return self._client @property @@ -89,9 +110,13 @@ def stream( return self._get_client().stream(method, endpoint, **kwargs) async def open(self) -> None: - if not self._client: - self._client = httpx.AsyncClient(**self._client_config) - await self._client.__aenter__() + """Enter a scope, opening the transport if it is not open yet.""" + async with self._lifecycle_lock: + self._scopes += 1 + if self._client is None: + client = httpx.AsyncClient(**self._client_config) + await client.__aenter__() + self._client = client async def close( self, @@ -99,9 +124,17 @@ async def close( exc_val: BaseException | None = None, exc_tb: TracebackType | None = None, ) -> None: - if self._client: - await self._client.__aexit__(exc_type, exc_val, exc_tb) - self._client = None + """Leave a scope, closing the transport once the last one exits.""" + async with self._lifecycle_lock: + if self._scopes > 0: + self._scopes -= 1 + if self._scopes > 0 or self._client is None: + return + client, self._client = self._client, None + + # Drain outside the lock so a slow teardown cannot block a + # concurrent open() of the next transport. + await client.__aexit__(exc_type, exc_val, exc_tb) async def __aenter__(self) -> "APIHttpClient": await self.open() @@ -118,8 +151,6 @@ async def __aexit__( async def request( self, method: str, endpoint: str, **kwargs: Any ) -> httpx.Response: - client = self._get_client() - log.debug(f"Request: {method} {endpoint}") # Never log bodies or headers: request payloads (e.g. env vars) # can contain secrets. @@ -137,8 +168,21 @@ async def request( for attempt in range(attempts): last_attempt = attempt == attempts - 1 + # Re-resolve every attempt: another task may have closed the + # transport while we were backing off. + client = self._get_client() try: response = await client.request(method, endpoint, **kwargs) + except RuntimeError as e: + # httpx raises a bare RuntimeError when the client is closed + # mid-flight. Anything else is not ours to interpret. + if self._client is not None: + raise + log.error(f"Client closed during {method} {endpoint}") + raise ClientStateError( + "The HTTP client was closed while the request to " + f"{endpoint} was in flight." + ) from e except httpx.TimeoutException as e: if not last_attempt: await self._sleep_before_retry(attempt, None, method, endpoint) @@ -166,6 +210,16 @@ async def request( f"for {method} {endpoint}" ) + if _is_delete_of_gone_resource(method, response.status_code, attempt): + # The first attempt reached the server and deleted the + # resource; only its response was lost. Reporting "not + # found" here would call a successful delete a failure. + log.debug( + f"Treating 404 on retried {method} {endpoint} as success: " + "the resource is gone, which is what was asked for" + ) + return response + if response.status_code in self._retry.retry_statuses and not last_attempt: await self._sleep_before_retry(attempt, response, method, endpoint) continue diff --git a/src/codesphere/_async/resources/team/domain/schemas.py b/src/codesphere/_async/resources/team/domain/schemas.py index c35cef0..ebe59fe 100644 --- a/src/codesphere/_async/resources/team/domain/schemas.py +++ b/src/codesphere/_async/resources/team/domain/schemas.py @@ -11,7 +11,6 @@ DomainVerificationStatus, RoutingMap, ) -from codesphere.utils import update_model_fields from ....core.base import BoundModel @@ -28,6 +27,17 @@ class Domain(DomainBase, BoundModel): + """A custom domain. + + Writes return the server's view of the domain and invalidate this + instance: the returned object is authoritative for that moment, while + ``self`` may have been overtaken by a concurrent write. Use the return + value, or call :meth:`refresh`. + """ + + # A domain is keyed by name within its team, not by a numeric id. + _identity_fields = ("name", "team_id") + async def update( self, data: CustomDomainConfig, @@ -39,9 +49,19 @@ async def update( response = await self._execute( _UPDATE_OP, team_id=self.team_id, name=self.name, data=data, timeout=timeout ) - update_model_fields(target=self, source=response) + self._mark_stale() return response + async def refresh(self, *, timeout: httpx.Timeout | float | None = None) -> Domain: + """Re-read this domain from the server, in place.""" + from .operations import _GET_OP + + fresh = await self._execute( + _GET_OP, team_id=self.team_id, name=self.name, timeout=timeout + ) + self._adopt(fresh) + return self + async def update_workspace_connections( self, connections: DomainRouting | RoutingMap, @@ -60,7 +80,7 @@ async def update_workspace_connections( data=payload, timeout=timeout, ) - update_model_fields(target=self, source=response) + self._mark_stale() return response async def verify_status( @@ -71,7 +91,9 @@ async def verify_status( response = await self._execute( _VERIFY_OP, team_id=self.team_id, name=self.name, timeout=timeout ) - update_model_fields(target=self.domain_verification_status, source=response) + # Triggers verification server-side, so this instance's nested + # status can be overtaken the same way an update can. + self._mark_stale() return response async def delete(self, *, timeout: httpx.Timeout | float | None = None) -> None: diff --git a/src/codesphere/_async/resources/workspace/landscape/resources.py b/src/codesphere/_async/resources/workspace/landscape/resources.py index e98e7b4..55bf587 100644 --- a/src/codesphere/_async/resources/workspace/landscape/resources.py +++ b/src/codesphere/_async/resources/workspace/landscape/resources.py @@ -4,10 +4,13 @@ import base64 import logging import re +import time from typing import TYPE_CHECKING import httpx +from codesphere.exceptions import ConflictError + from ....core.base import ResourceList, WorkspaceScopedResource from .operations import ( _DEPLOY_OP, @@ -22,6 +25,7 @@ from .schemas import ( PipelineStage, PipelineState, + PipelineStatus, PipelineStatusList, Profile, ProfileConfig, @@ -50,6 +54,32 @@ def _profile_filename(name: str) -> str: return f"ci.{name}.yml" +def _pin_run_identity( + pinned: dict[tuple[str, str], str], + statuses: list[PipelineStatus], + stage_name: str, +) -> None: + """Fail if the pipeline run being waited on was replaced by another. + + The API exposes no run id, but ``started_at`` changes when a stage is + restarted. First sighting pins it; a later different value means an + external actor redeployed and the original run is gone. + """ + for status in statuses: + if status.started_at is None: + continue + key = (status.server, status.replica) + first_seen = pinned.setdefault(key, status.started_at) + if first_seen != status.started_at: + raise ConflictError( + f"Pipeline stage '{stage_name}' on server '{status.server}' " + f"(replica '{status.replica}') was restarted while waiting: " + f"it began at {first_seen}, now {status.started_at}. Someone " + "else redeployed this workspace, so the run you were waiting " + "for no longer exists." + ) + + class WorkspaceLandscapeManager(WorkspaceScopedResource): async def _run_command( self, command: str, *, timeout: httpx.Timeout | float | None = None @@ -206,13 +236,24 @@ async def wait_for_stage( poll_interval: float = 5.0, server: str | None = None, ) -> PipelineStatusList: + """Poll until every relevant server finishes ``stage``. + + The run being waited on is pinned by its ``started_at``. If someone + else restarts the stage meanwhile — a colleague redeploying, a CI + job — this raises :class:`~codesphere.ConflictError` rather than + reporting the outcome of a run the caller never started. + + ``timeout`` is wall clock: time spent in the status requests counts + against it, not just the sleeps. + """ if poll_interval <= 0: raise ValueError("poll_interval must be greater than 0") stage_name = stage.value if isinstance(stage, PipelineStage) else stage - elapsed = 0.0 + deadline = time.monotonic() + timeout + pinned: dict[tuple[str, str], str] = {} - while elapsed < timeout: + while True: status_list = await self.get_stage_status(stage) relevant_statuses = [] @@ -224,35 +265,39 @@ async def wait_for_stage( if s.steps or s.state != PipelineState.WAITING: relevant_statuses.append(s) - if not relevant_statuses: + _pin_run_identity(pinned, relevant_statuses, stage_name) + + if relevant_statuses: + all_completed = all( + s.state + in ( + PipelineState.SUCCESS, + PipelineState.FAILURE, + PipelineState.ABORTED, + ) + for s in relevant_statuses + ) + if all_completed: + log.debug("Pipeline stage '%s' completed.", stage_name) + return PipelineStatusList(root=relevant_statuses) + + states = [f"{s.server}={s.state.value}" for s in relevant_statuses] + log.debug( + "Pipeline stage '%s' status: %s (%.1fs remaining)", + stage_name, + ", ".join(states), + max(0.0, deadline - time.monotonic()), + ) + else: log.debug( "Pipeline stage '%s': no servers with steps yet, waiting...", stage_name, ) - await asyncio.sleep(poll_interval) - elapsed += poll_interval - continue - - all_completed = all( - s.state - in (PipelineState.SUCCESS, PipelineState.FAILURE, PipelineState.ABORTED) - for s in relevant_statuses - ) - - if all_completed: - log.debug("Pipeline stage '%s' completed.", stage_name) - return PipelineStatusList(root=relevant_statuses) - states = [f"{s.server}={s.state.value}" for s in relevant_statuses] - log.debug( - "Pipeline stage '%s' status: %s (elapsed: %.1fs)", - stage_name, - ", ".join(states), - elapsed, - ) - await asyncio.sleep(poll_interval) - elapsed += poll_interval - - raise TimeoutError( - f"Pipeline stage '{stage_name}' did not complete within {timeout} seconds." - ) + remaining = deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError( + f"Pipeline stage '{stage_name}' did not complete within " + f"{timeout} seconds." + ) + await asyncio.sleep(min(poll_interval, remaining)) diff --git a/src/codesphere/_async/resources/workspace/logs/resources.py b/src/codesphere/_async/resources/workspace/logs/resources.py index 83e41fb..b9df3da 100644 --- a/src/codesphere/_async/resources/workspace/logs/resources.py +++ b/src/codesphere/_async/resources/workspace/logs/resources.py @@ -10,7 +10,12 @@ import httpx from codesphere.core.operations import StreamOperation -from codesphere.exceptions import APIError, ValidationError, raise_for_status +from codesphere.exceptions import ( + APIError, + ClientStateError, + ValidationError, + raise_for_status, +) from ....core.base import WorkspaceScopedResource from ....http_client import APIHttpClient @@ -58,7 +63,12 @@ def _deprecated(old: str, new: str) -> None: class LogStream: - """Async context manager for streaming logs via SSE.""" + """Async context manager for streaming logs via SSE. + + Single-use and single-consumer: an SSE response body can only be read + once. Call ``logs.open(...)`` again for another stream rather than + re-entering or iterating this one twice. + """ def __init__( self, @@ -73,8 +83,17 @@ def __init__( self._timeout = timeout self._response: httpx.Response | None = None self._stream_context: Any = None + self._entered = False + self._consumer_started = False async def __aenter__(self) -> LogStream: + if self._entered: + # Overwriting _stream_context here would strand the first one. + raise ClientStateError( + "This LogStream has already been opened. Streams are " + "single-use: call logs.open(...) again for a new one." + ) + self._entered = True headers = {"Accept": "text/event-stream"} # A user timeout bounds the read timeout too, so a silent stream # wakes up instead of blocking past the deadline. @@ -101,6 +120,14 @@ async def __aexit__(self, exc_type, exc_val, exc_tb) -> None: await self._stream_context.__aexit__(exc_type, exc_val, exc_tb) def __aiter__(self) -> AsyncIterator[LogEntry]: + if self._consumer_started: + # Two iterators over one response interleave and corrupt the + # SSE frame parsing; httpx would raise StreamConsumed later. + raise ClientStateError( + "This LogStream is already being consumed. Streams have a " + "single consumer: call logs.open(...) again for a new one." + ) + self._consumer_started = True return self._iterate() async def _iterate(self) -> AsyncIterator[LogEntry]: diff --git a/src/codesphere/_async/resources/workspace/schemas.py b/src/codesphere/_async/resources/workspace/schemas.py index 27e8f15..83d3c7d 100644 --- a/src/codesphere/_async/resources/workspace/schemas.py +++ b/src/codesphere/_async/resources/workspace/schemas.py @@ -2,6 +2,7 @@ import asyncio import logging +import time from functools import cached_property import httpx @@ -14,7 +15,6 @@ WorkspaceStatus, WorkspaceUpdate, ) -from codesphere.utils import update_model_fields from ...core.base import BoundModel from .env_vars import EnvVar, WorkspaceEnvVarManager @@ -43,10 +43,29 @@ async def update( *, timeout: httpx.Timeout | float | None = None, ) -> None: + """Apply a partial update, then invalidate this instance. + + The PATCH returns no body, and the platform orders writes by + arrival while this client sees them by response, so the resulting + server state is not knowable from here. Rather than write the + request payload back and risk reporting a value the platform does + not hold, the instance is marked stale: reads raise + :class:`~codesphere.StaleModelError` until :meth:`refresh`. + """ from .operations import _UPDATE_OP await self._execute(_UPDATE_OP, id=self.id, data=data, timeout=timeout) - update_model_fields(target=self, source=data) + self._mark_stale() + + async def refresh( + self, *, timeout: httpx.Timeout | float | None = None + ) -> Workspace: + """Re-read this workspace from the server, in place.""" + from .operations import _GET_OP + + fresh = await self._execute(_GET_OP, workspace_id=self.id, timeout=timeout) + self._adopt(fresh) + return self async def delete(self, *, timeout: httpx.Timeout | float | None = None) -> None: from .operations import _DELETE_OP @@ -69,24 +88,28 @@ async def wait_until_running( if poll_interval <= 0: raise ValueError("poll_interval must be greater than 0") - elapsed = 0.0 - while elapsed < timeout: + # Wall-clock deadline: time spent in the status requests counts + # against the timeout, not just the sleeps between them. + deadline = time.monotonic() + timeout + while True: status = await self.get_status() if status.is_running: log.debug("Workspace %s is now running.", self.id) return + + remaining = deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError( + f"Workspace {self.id} did not reach running state within " + f"{timeout} seconds." + ) log.debug( - "Workspace %s not running yet, waiting %ss... (elapsed: %.1fs)", + "Workspace %s not running yet, waiting %ss... (%.1fs remaining)", self.id, poll_interval, - elapsed, + remaining, ) - await asyncio.sleep(poll_interval) - elapsed += poll_interval - - raise TimeoutError( - f"Workspace {self.id} did not reach running state within {timeout} seconds." - ) + await asyncio.sleep(min(poll_interval, remaining)) async def execute_command( self, diff --git a/src/codesphere/_sync/core/base.py b/src/codesphere/_sync/core/base.py index 95e02b2..3411fe2 100644 --- a/src/codesphere/_sync/core/base.py +++ b/src/codesphere/_sync/core/base.py @@ -1,7 +1,7 @@ # Do not edit this file directly. It has been autogenerated from # src/codesphere/_async/core/base.py from collections.abc import Mapping -from typing import Any, TypeVar +from typing import Any, ClassVar, TypeVar import httpx from pydantic import PrivateAttr @@ -9,6 +9,7 @@ from codesphere.core.models import CamelModel from codesphere.core.models import ResourceList as ResourceList from codesphere.core.operations import APIOperation +from codesphere.exceptions import StaleModelError from codesphere.feature_flags import FlagRequirement from ..http_client import APIHttpClient @@ -68,9 +69,78 @@ class BoundModel(CamelModel): Instances returned by the SDK get their client attached automatically, which lets entity methods (e.g. ``workspace.delete()``) make further API calls. + + A write that the SDK cannot verify marks the instance **stale**: its + field values are dropped and reads raise + :class:`~codesphere.StaleModelError` until ``refresh()`` re-reads the + entity. Identity fields survive, so the instance can still be logged + and re-fetched. See :meth:`_mark_stale`. """ + #: Fields that stay readable on a stale instance. An entity's identity + #: cannot be changed by a write, so it is always safe to report. + _identity_fields: ClassVar[tuple[str, ...]] = ("id",) + _http_client: APIHttpClient | None = PrivateAttr(default=None) + _stale: bool = PrivateAttr(default=False) + + def __getattr__(self, item: str) -> Any: + # Pydantic keeps field values in __dict__, so this only runs once a + # lookup has already failed: zero cost until a model goes stale. + if self._is_stale() and item in type(self).__pydantic_fields__: + raise StaleModelError(type(self).__name__, item) + # Delegate to pydantic, which resolves private attributes here. + # It defines __getattr__ only at runtime, so it is fetched + # dynamically rather than called through super() directly. + parent = getattr(super(), "__getattr__", None) + if parent is None: # pragma: no cover - pydantic always defines it + raise AttributeError(item) + return parent(item) + + def _is_stale(self) -> bool: + private = object.__getattribute__(self, "__pydantic_private__") + return bool(private and private.get("_stale")) + + def _mark_stale(self) -> None: + """Drop local field values the SDK can no longer vouch for. + + Called after a write whose resulting server state is unknown. + Clearing ``__dict__`` (rather than setting a flag beside intact + values) is what routes later reads through ``__getattr__``. + """ + identity = { + name: value + for name, value in self.__dict__.items() + if name in type(self)._identity_fields + } + # Also drops cached_property managers; they rebuild after refresh. + self.__dict__.clear() + self.__dict__.update(identity) + self._stale = True + + def _adopt(self, fresh: "BoundModel") -> None: + """Repopulate from a server-authoritative re-read.""" + self.__dict__.clear() + self.__dict__.update(fresh.__dict__) + self._stale = False + + def model_dump(self, *args: Any, **kwargs: Any) -> dict[str, Any]: + # Pydantic serializes straight from __dict__, so without this a + # stale model would quietly dump only its identity fields. + if self._is_stale(): + raise StaleModelError(type(self).__name__) + return super().model_dump(*args, **kwargs) + + def model_dump_json(self, *args: Any, **kwargs: Any) -> str: + if self._is_stale(): + raise StaleModelError(type(self).__name__) + return super().model_dump_json(*args, **kwargs) + + def __repr_args__(self) -> Any: + # Keep repr() working for debugging, but say why it looks empty. + if self._is_stale(): + return [*super().__repr_args__(), ("stale", True)] + return super().__repr_args__() def _client(self) -> APIHttpClient: if self._http_client is None or not hasattr(self._http_client, "request"): diff --git a/src/codesphere/_sync/flags_store.py b/src/codesphere/_sync/flags_store.py index 9073730..79fe946 100644 --- a/src/codesphere/_sync/flags_store.py +++ b/src/codesphere/_sync/flags_store.py @@ -29,6 +29,11 @@ class FlagsStore: client's lifetime. A 404 — the instance predates feature flags — is cached as an empty snapshot: reads report nothing available and gated calls fail closed. Transient errors propagate uncached. + + Safe to share across tasks: the snapshot and the ``legacy_platform`` + flag it was fetched with are always read as one pair, and a fetch + that was in flight when :meth:`invalidate` ran does not repopulate + the cache it was told to drop. """ def __init__(self, client: "APIHttpClient") -> None: @@ -36,25 +41,41 @@ def __init__(self, client: "APIHttpClient") -> None: self._snapshot: FlagsSnapshot | None = None self._legacy_platform = False self._lock = Lock() + # Bumped by invalidate(). A fetch stamps the generation it started + # under and refuses to install its result if that has moved on. + self._generation = 0 def get(self, *, refresh: bool = False) -> FlagsSnapshot: + snapshot, _ = self._get_with_legacy(refresh=refresh) + return snapshot + + def _get_with_legacy(self, *, refresh: bool = False) -> tuple[FlagsSnapshot, bool]: + """The snapshot and the legacy-platform flag it was fetched with. + + Returned together from one lock acquisition so callers cannot pair + a snapshot with a legacy flag from a different fetch. + """ with self._lock: - if self._snapshot is None or refresh: - self._snapshot = self._fetch() - return self._snapshot + if self._snapshot is not None and not refresh: + return self._snapshot, self._legacy_platform - def _fetch(self) -> FlagsSnapshot: + generation = self._generation + snapshot, legacy = self._fetch() + if generation == self._generation: + self._snapshot = snapshot + self._legacy_platform = legacy + return snapshot, legacy + + def _fetch(self) -> tuple[FlagsSnapshot, bool]: try: response = self._client.request("GET", FLAGS_ENDPOINT) except NotFoundError: - self._legacy_platform = True - return FlagsSnapshot.empty() - self._legacy_platform = False - return FlagsSnapshot.model_validate(response.json()) + return FlagsSnapshot.empty(), True + return FlagsSnapshot.model_validate(response.json()), False def require(self, requirement: FlagRequirement) -> None: """Raise unless the required flag is enabled on this instance.""" - snapshot = self.get() + snapshot, legacy_platform = self._get_with_legacy() flag, category = requirement.flag, requirement.category category_str = str(category) if category is not None else None @@ -62,7 +83,7 @@ def require(self, requirement: FlagRequirement) -> None: raise FeatureNotAvailableError( flag=flag, category=category_str, - legacy_platform=self._legacy_platform, + legacy_platform=legacy_platform, ) if not snapshot.is_enabled(flag, category): raise FeatureNotEnabledError(flag=flag, category=category_str) @@ -72,5 +93,11 @@ def cached(self) -> FlagsSnapshot | None: return self._snapshot def invalidate(self) -> None: - """Drop the cached snapshot; the next need re-fetches.""" + """Drop the cached snapshot; the next need re-fetches. + + Stays synchronous on purpose: acquiring the lock here would block + the caller for as long as an in-flight flags request takes. The + generation bump is what makes that safe. + """ + self._generation += 1 self._snapshot = None diff --git a/src/codesphere/_sync/http_client.py b/src/codesphere/_sync/http_client.py index 46edbb9..691b972 100644 --- a/src/codesphere/_sync/http_client.py +++ b/src/codesphere/_sync/http_client.py @@ -13,8 +13,14 @@ from pydantic import SecretStr from codesphere.config import RetryConfig -from codesphere.exceptions import NetworkError, TimeoutError, raise_for_status - +from codesphere.exceptions import ( + ClientStateError, + NetworkError, + TimeoutError, + raise_for_status, +) + +from ._compat import Lock from .flags_store import FlagsStore import time @@ -34,12 +40,25 @@ def _parse_retry_after(value: str | None) -> float | None: return max(0.0, (retry_at - datetime.now(UTC)).total_seconds()) +def _is_delete_of_gone_resource(method: str, status_code: int, attempt: int) -> bool: + """Whether a 404 means "the DELETE already succeeded" rather than a miss. + + Only ever true after at least one retry: a 404 on the very first + attempt is a genuine "no such resource" and must still raise. + """ + return attempt > 0 and status_code == 404 and method.upper() == "DELETE" + + class APIHttpClient: """Thin transport around httpx. Receives its full configuration from the caller (the SDK resolves arguments, environment variables, and defaults) and owns nothing but the connection lifecycle, error mapping, and opt-in retries. + + Safe to share across tasks. ``open()``/``close()`` are reference + counted, so nested and concurrent scopes each hold the transport open + and only the last one to exit tears it down. """ def __init__( @@ -54,6 +73,11 @@ def __init__( self._base_url = base_url self._client: httpx.Client | None = None self._retry = retry if retry is not None else RetryConfig() + # Guards the check-then-act in open()/close(). Without it, two + # threads in the sync twin can each build a transport and orphan + # one of them. + self._lifecycle_lock = Lock() + self._scopes = 0 # Per-connection feature-flags cache; the transport never reads it # itself (see feature_flags.FlagsStore). self.flags_store = FlagsStore(self) @@ -67,10 +91,7 @@ def __init__( def _get_client(self) -> httpx.Client: if not self._client: - raise RuntimeError( - "Client is not open. Use the client as a context manager " - "or call open() before making requests." - ) + raise ClientStateError() return self._client @property @@ -92,9 +113,13 @@ def stream( return self._get_client().stream(method, endpoint, **kwargs) def open(self) -> None: - if not self._client: - self._client = httpx.Client(**self._client_config) - self._client.__enter__() + """Enter a scope, opening the transport if it is not open yet.""" + with self._lifecycle_lock: + self._scopes += 1 + if self._client is None: + client = httpx.Client(**self._client_config) + client.__enter__() + self._client = client def close( self, @@ -102,9 +127,17 @@ def close( exc_val: BaseException | None = None, exc_tb: TracebackType | None = None, ) -> None: - if self._client: - self._client.__exit__(exc_type, exc_val, exc_tb) - self._client = None + """Leave a scope, closing the transport once the last one exits.""" + with self._lifecycle_lock: + if self._scopes > 0: + self._scopes -= 1 + if self._scopes > 0 or self._client is None: + return + client, self._client = self._client, None + + # Drain outside the lock so a slow teardown cannot block a + # concurrent open() of the next transport. + client.__exit__(exc_type, exc_val, exc_tb) def __enter__(self) -> "APIHttpClient": self.open() @@ -119,8 +152,6 @@ def __exit__( self.close(exc_type, exc_val, exc_tb) def request(self, method: str, endpoint: str, **kwargs: Any) -> httpx.Response: - client = self._get_client() - log.debug(f"Request: {method} {endpoint}") # Never log bodies or headers: request payloads (e.g. env vars) # can contain secrets. @@ -138,8 +169,21 @@ def request(self, method: str, endpoint: str, **kwargs: Any) -> httpx.Response: for attempt in range(attempts): last_attempt = attempt == attempts - 1 + # Re-resolve every attempt: another task may have closed the + # transport while we were backing off. + client = self._get_client() try: response = client.request(method, endpoint, **kwargs) + except RuntimeError as e: + # httpx raises a bare RuntimeError when the client is closed + # mid-flight. Anything else is not ours to interpret. + if self._client is not None: + raise + log.error(f"Client closed during {method} {endpoint}") + raise ClientStateError( + "The HTTP client was closed while the request to " + f"{endpoint} was in flight." + ) from e except httpx.TimeoutException as e: if not last_attempt: self._sleep_before_retry(attempt, None, method, endpoint) @@ -167,6 +211,16 @@ def request(self, method: str, endpoint: str, **kwargs: Any) -> httpx.Response: f"for {method} {endpoint}" ) + if _is_delete_of_gone_resource(method, response.status_code, attempt): + # The first attempt reached the server and deleted the + # resource; only its response was lost. Reporting "not + # found" here would call a successful delete a failure. + log.debug( + f"Treating 404 on retried {method} {endpoint} as success: " + "the resource is gone, which is what was asked for" + ) + return response + if response.status_code in self._retry.retry_statuses and not last_attempt: self._sleep_before_retry(attempt, response, method, endpoint) continue diff --git a/src/codesphere/_sync/resources/team/domain/schemas.py b/src/codesphere/_sync/resources/team/domain/schemas.py index 163d960..fa1b0cf 100644 --- a/src/codesphere/_sync/resources/team/domain/schemas.py +++ b/src/codesphere/_sync/resources/team/domain/schemas.py @@ -13,7 +13,6 @@ DomainVerificationStatus, RoutingMap, ) -from codesphere.utils import update_model_fields from ....core.base import BoundModel @@ -30,6 +29,17 @@ class Domain(DomainBase, BoundModel): + """A custom domain. + + Writes return the server's view of the domain and invalidate this + instance: the returned object is authoritative for that moment, while + ``self`` may have been overtaken by a concurrent write. Use the return + value, or call :meth:`refresh`. + """ + + # A domain is keyed by name within its team, not by a numeric id. + _identity_fields = ("name", "team_id") + def update( self, data: CustomDomainConfig, @@ -41,9 +51,19 @@ def update( response = self._execute( _UPDATE_OP, team_id=self.team_id, name=self.name, data=data, timeout=timeout ) - update_model_fields(target=self, source=response) + self._mark_stale() return response + def refresh(self, *, timeout: httpx.Timeout | float | None = None) -> Domain: + """Re-read this domain from the server, in place.""" + from .operations import _GET_OP + + fresh = self._execute( + _GET_OP, team_id=self.team_id, name=self.name, timeout=timeout + ) + self._adopt(fresh) + return self + def update_workspace_connections( self, connections: DomainRouting | RoutingMap, @@ -62,7 +82,7 @@ def update_workspace_connections( data=payload, timeout=timeout, ) - update_model_fields(target=self, source=response) + self._mark_stale() return response def verify_status( @@ -73,7 +93,9 @@ def verify_status( response = self._execute( _VERIFY_OP, team_id=self.team_id, name=self.name, timeout=timeout ) - update_model_fields(target=self.domain_verification_status, source=response) + # Triggers verification server-side, so this instance's nested + # status can be overtaken the same way an update can. + self._mark_stale() return response def delete(self, *, timeout: httpx.Timeout | float | None = None) -> None: diff --git a/src/codesphere/_sync/resources/workspace/landscape/resources.py b/src/codesphere/_sync/resources/workspace/landscape/resources.py index c725d05..2b62117 100644 --- a/src/codesphere/_sync/resources/workspace/landscape/resources.py +++ b/src/codesphere/_sync/resources/workspace/landscape/resources.py @@ -6,10 +6,13 @@ import base64 import logging import re +import time from typing import TYPE_CHECKING import httpx +from codesphere.exceptions import ConflictError + from ....core.base import ResourceList, WorkspaceScopedResource from .operations import ( _DEPLOY_OP, @@ -24,11 +27,11 @@ from .schemas import ( PipelineStage, PipelineState, + PipelineStatus, PipelineStatusList, Profile, ProfileConfig, ) -import time if TYPE_CHECKING: from ..schemas import CommandOutput @@ -53,6 +56,32 @@ def _profile_filename(name: str) -> str: return f"ci.{name}.yml" +def _pin_run_identity( + pinned: dict[tuple[str, str], str], + statuses: list[PipelineStatus], + stage_name: str, +) -> None: + """Fail if the pipeline run being waited on was replaced by another. + + The API exposes no run id, but ``started_at`` changes when a stage is + restarted. First sighting pins it; a later different value means an + external actor redeployed and the original run is gone. + """ + for status in statuses: + if status.started_at is None: + continue + key = (status.server, status.replica) + first_seen = pinned.setdefault(key, status.started_at) + if first_seen != status.started_at: + raise ConflictError( + f"Pipeline stage '{stage_name}' on server '{status.server}' " + f"(replica '{status.replica}') was restarted while waiting: " + f"it began at {first_seen}, now {status.started_at}. Someone " + "else redeployed this workspace, so the run you were waiting " + "for no longer exists." + ) + + class WorkspaceLandscapeManager(WorkspaceScopedResource): def _run_command( self, command: str, *, timeout: httpx.Timeout | float | None = None @@ -203,13 +232,24 @@ def wait_for_stage( poll_interval: float = 5.0, server: str | None = None, ) -> PipelineStatusList: + """Poll until every relevant server finishes ``stage``. + + The run being waited on is pinned by its ``started_at``. If someone + else restarts the stage meanwhile — a colleague redeploying, a CI + job — this raises :class:`~codesphere.ConflictError` rather than + reporting the outcome of a run the caller never started. + + ``timeout`` is wall clock: time spent in the status requests counts + against it, not just the sleeps. + """ if poll_interval <= 0: raise ValueError("poll_interval must be greater than 0") stage_name = stage.value if isinstance(stage, PipelineStage) else stage - elapsed = 0.0 + deadline = time.monotonic() + timeout + pinned: dict[tuple[str, str], str] = {} - while elapsed < timeout: + while True: status_list = self.get_stage_status(stage) relevant_statuses = [] @@ -221,35 +261,39 @@ def wait_for_stage( if s.steps or s.state != PipelineState.WAITING: relevant_statuses.append(s) - if not relevant_statuses: + _pin_run_identity(pinned, relevant_statuses, stage_name) + + if relevant_statuses: + all_completed = all( + s.state + in ( + PipelineState.SUCCESS, + PipelineState.FAILURE, + PipelineState.ABORTED, + ) + for s in relevant_statuses + ) + if all_completed: + log.debug("Pipeline stage '%s' completed.", stage_name) + return PipelineStatusList(root=relevant_statuses) + + states = [f"{s.server}={s.state.value}" for s in relevant_statuses] + log.debug( + "Pipeline stage '%s' status: %s (%.1fs remaining)", + stage_name, + ", ".join(states), + max(0.0, deadline - time.monotonic()), + ) + else: log.debug( "Pipeline stage '%s': no servers with steps yet, waiting...", stage_name, ) - time.sleep(poll_interval) - elapsed += poll_interval - continue - - all_completed = all( - s.state - in (PipelineState.SUCCESS, PipelineState.FAILURE, PipelineState.ABORTED) - for s in relevant_statuses - ) - - if all_completed: - log.debug("Pipeline stage '%s' completed.", stage_name) - return PipelineStatusList(root=relevant_statuses) - states = [f"{s.server}={s.state.value}" for s in relevant_statuses] - log.debug( - "Pipeline stage '%s' status: %s (elapsed: %.1fs)", - stage_name, - ", ".join(states), - elapsed, - ) - time.sleep(poll_interval) - elapsed += poll_interval - - raise TimeoutError( - f"Pipeline stage '{stage_name}' did not complete within {timeout} seconds." - ) + remaining = deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError( + f"Pipeline stage '{stage_name}' did not complete within " + f"{timeout} seconds." + ) + time.sleep(min(poll_interval, remaining)) diff --git a/src/codesphere/_sync/resources/workspace/logs/resources.py b/src/codesphere/_sync/resources/workspace/logs/resources.py index d1fd677..a0f3355 100644 --- a/src/codesphere/_sync/resources/workspace/logs/resources.py +++ b/src/codesphere/_sync/resources/workspace/logs/resources.py @@ -12,7 +12,12 @@ import httpx from codesphere.core.operations import StreamOperation -from codesphere.exceptions import APIError, ValidationError, raise_for_status +from codesphere.exceptions import ( + APIError, + ClientStateError, + ValidationError, + raise_for_status, +) from ....core.base import WorkspaceScopedResource from ....http_client import APIHttpClient @@ -60,7 +65,12 @@ def _deprecated(old: str, new: str) -> None: class LogStream: - """Async context manager for streaming logs via SSE.""" + """Async context manager for streaming logs via SSE. + + Single-use and single-consumer: an SSE response body can only be read + once. Call ``logs.open(...)`` again for another stream rather than + re-entering or iterating this one twice. + """ def __init__( self, @@ -75,8 +85,17 @@ def __init__( self._timeout = timeout self._response: httpx.Response | None = None self._stream_context: Any = None + self._entered = False + self._consumer_started = False def __enter__(self) -> LogStream: + if self._entered: + # Overwriting _stream_context here would strand the first one. + raise ClientStateError( + "This LogStream has already been opened. Streams are " + "single-use: call logs.open(...) again for a new one." + ) + self._entered = True headers = {"Accept": "text/event-stream"} # A user timeout bounds the read timeout too, so a silent stream # wakes up instead of blocking past the deadline. @@ -103,6 +122,14 @@ def __exit__(self, exc_type, exc_val, exc_tb) -> None: self._stream_context.__exit__(exc_type, exc_val, exc_tb) def __iter__(self) -> Iterator[LogEntry]: + if self._consumer_started: + # Two iterators over one response interleave and corrupt the + # SSE frame parsing; httpx would raise StreamConsumed later. + raise ClientStateError( + "This LogStream is already being consumed. Streams have a " + "single consumer: call logs.open(...) again for a new one." + ) + self._consumer_started = True return self._iterate() def _iterate(self) -> Iterator[LogEntry]: diff --git a/src/codesphere/_sync/resources/workspace/schemas.py b/src/codesphere/_sync/resources/workspace/schemas.py index 6a5089b..484bf60 100644 --- a/src/codesphere/_sync/resources/workspace/schemas.py +++ b/src/codesphere/_sync/resources/workspace/schemas.py @@ -4,6 +4,7 @@ import asyncio import logging +import time from functools import cached_property import httpx @@ -16,14 +17,12 @@ WorkspaceStatus, WorkspaceUpdate, ) -from codesphere.utils import update_model_fields from ...core.base import BoundModel from .env_vars import EnvVar, WorkspaceEnvVarManager from .git import WorkspaceGitManager from .landscape import WorkspaceLandscapeManager from .logs import WorkspaceLogManager -import time log = logging.getLogger(__name__) @@ -46,10 +45,27 @@ def update( *, timeout: httpx.Timeout | float | None = None, ) -> None: + """Apply a partial update, then invalidate this instance. + + The PATCH returns no body, and the platform orders writes by + arrival while this client sees them by response, so the resulting + server state is not knowable from here. Rather than write the + request payload back and risk reporting a value the platform does + not hold, the instance is marked stale: reads raise + :class:`~codesphere.StaleModelError` until :meth:`refresh`. + """ from .operations import _UPDATE_OP self._execute(_UPDATE_OP, id=self.id, data=data, timeout=timeout) - update_model_fields(target=self, source=data) + self._mark_stale() + + def refresh(self, *, timeout: httpx.Timeout | float | None = None) -> Workspace: + """Re-read this workspace from the server, in place.""" + from .operations import _GET_OP + + fresh = self._execute(_GET_OP, workspace_id=self.id, timeout=timeout) + self._adopt(fresh) + return self def delete(self, *, timeout: httpx.Timeout | float | None = None) -> None: from .operations import _DELETE_OP @@ -72,24 +88,28 @@ def wait_until_running( if poll_interval <= 0: raise ValueError("poll_interval must be greater than 0") - elapsed = 0.0 - while elapsed < timeout: + # Wall-clock deadline: time spent in the status requests counts + # against the timeout, not just the sleeps between them. + deadline = time.monotonic() + timeout + while True: status = self.get_status() if status.is_running: log.debug("Workspace %s is now running.", self.id) return + + remaining = deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError( + f"Workspace {self.id} did not reach running state within " + f"{timeout} seconds." + ) log.debug( - "Workspace %s not running yet, waiting %ss... (elapsed: %.1fs)", + "Workspace %s not running yet, waiting %ss... (%.1fs remaining)", self.id, poll_interval, - elapsed, + remaining, ) - time.sleep(poll_interval) - elapsed += poll_interval - - raise TimeoutError( - f"Workspace {self.id} did not reach running state within {timeout} seconds." - ) + time.sleep(min(poll_interval, remaining)) def execute_command( self, diff --git a/src/codesphere/config.py b/src/codesphere/config.py index 6c3615f..4179987 100644 --- a/src/codesphere/config.py +++ b/src/codesphere/config.py @@ -17,6 +17,23 @@ class RetryConfig: otherwise exponential backoff with jitter is used. Only idempotent methods are retried by default; add ``"POST"`` to ``retry_methods`` explicitly if your endpoints tolerate it. + + Retries and duplicate side effects + ---------------------------------- + ``PUT`` and ``DELETE`` are idempotent as HTTP methods, but not always + in effect. When the first attempt reaches the platform and only its + response is lost (a gateway 503, a dropped connection), the retry runs + the operation a second time — a landscape teardown, for example, can + execute twice. The Codesphere API has no idempotency keys, so the SDK + cannot deduplicate this for you. + + The SDK does handle the most common consequence: a 404 on a retried + ``DELETE`` is treated as success, because the resource being gone is + what the caller asked for. A 404 on the *first* attempt still raises + :class:`~codesphere.NotFoundError`. + + If duplicate execution is unacceptable for your workload, set + ``max_retries=0`` or drop the method from ``retry_methods``. """ max_retries: int = 2 diff --git a/src/codesphere/exceptions.py b/src/codesphere/exceptions.py index a3d520e..672ea9c 100644 --- a/src/codesphere/exceptions.py +++ b/src/codesphere/exceptions.py @@ -160,6 +160,45 @@ def __init__(self, message: str | None = None): super().__init__(message) +class StaleModelError(CodesphereError): + """Raised when reading a model whose local state is no longer trustworthy. + + A write succeeded, but the SDK cannot know the resulting server state: + the platform orders writes by arrival while the client sees them by + response, and anyone else may have written in between. Rather than + report a value the platform may not hold, the model refuses to answer + until ``refresh()`` re-reads it. + """ + + def __init__(self, model: str, field: str | None = None): + self.model = model + self.field = field + target = f"{model}.{field}" if field else model + super().__init__( + f"Cannot read {target}: this {model} was invalidated by a write " + "and its local values may disagree with the platform. " + "Call `await refresh()` to re-read it from the server." + ) + + +class ClientStateError(CodesphereError, RuntimeError): + """Raised when the HTTP client is not usable for the attempted request. + + Either it was never opened, or it was closed by another task while the + request was in flight. Also inherits from :class:`RuntimeError`, which + is what the SDK raised for this before it joined the exception + hierarchy, so existing ``except RuntimeError`` handlers keep working. + """ + + def __init__(self, message: str | None = None): + if message is None: + message = ( + "Client is not open. Use the client as a context manager " + "or call open() before making requests." + ) + super().__init__(message) + + def _flag_label(flag: str, category: str | None) -> str: if category: return f"'{flag}' (category '{category}')" diff --git a/src/codesphere/utils.py b/src/codesphere/utils.py deleted file mode 100644 index 8e00a9d..0000000 --- a/src/codesphere/utils.py +++ /dev/null @@ -1,16 +0,0 @@ -import logging - -from pydantic import BaseModel - -log = logging.getLogger(__name__) - - -def update_model_fields(target: BaseModel, source: BaseModel) -> None: - if log.isEnabledFor(logging.DEBUG): - # Field names only — values may contain secrets (e.g. env vars). - fields = ", ".join(sorted(source.model_fields_set)) - log.debug(f"Updating {target.__class__.__name__} fields: {fields}") - - for field_name in source.model_fields_set: - value = getattr(source, field_name) - setattr(target, field_name, value) diff --git a/tests/resources/workspace/test_workspace.py b/tests/resources/workspace/test_workspace.py index 994eaba..32b4ba5 100644 --- a/tests/resources/workspace/test_workspace.py +++ b/tests/resources/workspace/test_workspace.py @@ -1,5 +1,6 @@ import pytest +from codesphere.exceptions import StaleModelError from codesphere.resources.workspace import ( Workspace, WorkspaceCreate, @@ -83,15 +84,30 @@ class TestWorkspaceModel: @pytest.mark.asyncio async def test_update_workspace(self, workspace_model_factory): - """Workspace.update() should update the workspace and local model.""" + """Workspace.update() sends the PATCH and invalidates the model.""" workspace, mock_client = workspace_model_factory() update_data = WorkspaceUpdate(name="updated-name", plan_id=10) await workspace.update(data=update_data) mock_client.request.assert_awaited_once() - assert workspace.name == "updated-name" - assert workspace.plan_id == 10 + # The PATCH has no body and the server may have been written to by + # someone else, so the local values are no longer trustworthy. + with pytest.raises(StaleModelError): + _ = workspace.name + with pytest.raises(StaleModelError): + workspace.to_dict() + + @pytest.mark.asyncio + async def test_identity_survives_invalidation(self, workspace_model_factory): + """The id is readable on a stale model: a write cannot change it.""" + workspace, _ = workspace_model_factory() + original_id = workspace.id + + await workspace.update(data=WorkspaceUpdate(name="updated-name")) + + assert workspace.id == original_id + assert "stale" in repr(workspace) @pytest.mark.asyncio async def test_delete_workspace(self, workspace_model_factory): diff --git a/tests/sync/test_concurrency.py b/tests/sync/test_concurrency.py new file mode 100644 index 0000000..e779528 --- /dev/null +++ b/tests/sync/test_concurrency.py @@ -0,0 +1,168 @@ +"""Concurrency regression tests for the generated sync client. + +The sync twin runs under real threads, so check-then-act patterns that are +invisible in asyncio (no await between the check and the assignment) become +genuine data races here. ``threading.Barrier`` widens the window +deterministically instead of relying on scheduling luck. +""" + +import threading +import time + +import httpx +import pytest +import respx + +from codesphere import RetryConfig +from codesphere.exceptions import CodesphereError +from codesphere.feature_flags import FLAGS_ENDPOINT +from codesphere.sync import CodesphereSDK + +BASE = "https://sync-concurrency.test/api" + +FLAGS_JSON = {"features": {"available": ["gadgets"], "enabled": ["gadgets"]}} + + +@pytest.fixture +def api(): + with respx.mock(base_url=BASE, assert_all_called=False) as router: + yield router + + +@pytest.fixture +def sdk(): + return CodesphereSDK(token="t", base_url=BASE, retry=RetryConfig(max_retries=0)) + + +def _run_in_threads(target, count: int) -> list[BaseException]: + errors: list[BaseException] = [] + + def wrapper(): + try: + target() + except BaseException as exc: # the assertion below surfaces it + errors.append(exc) + + threads = [threading.Thread(target=wrapper) for _ in range(count)] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=5) + return errors + + +class TestClientLifecycle: + def test_concurrent_open_builds_exactly_one_transport(self, api, sdk): + """Two threads racing open() must not each construct (and leak) a + connection pool.""" + constructed: list[httpx.Client] = [] + real_init = httpx.Client.__init__ + # Rendezvous before open(), so both threads enter it together. The + # barrier must not live inside the construction path: once the race + # is fixed only one thread constructs, and a barrier there would + # deadlock on the correct behavior. + start = threading.Barrier(2, timeout=5) + + def spy_init(self, *args, **kwargs): + constructed.append(self) + time.sleep(0.05) # hold the check-then-act window open + return real_init(self, *args, **kwargs) + + def open_together(): + start.wait() + sdk.open() + + with pytest.MonkeyPatch.context() as monkeypatch: + monkeypatch.setattr(httpx.Client, "__init__", spy_init) + errors = _run_in_threads(open_together, 2) + + try: + assert not errors, f"open() raised under contention: {errors}" + orphans = [c for c in constructed if c is not sdk._http_client._client] + assert len(constructed) == 1, ( + f"{len(constructed)} transports constructed, " + f"{len(orphans)} orphaned and never closed" + ) + finally: + # Both threads entered a scope; balance both. + sdk.close() + sdk.close() + + def test_concurrent_scopes_do_not_close_each_other(self, api, sdk): + api.get("/teams").respond(200, json=[]) + + second_entered = threading.Event() + first_exited = threading.Event() + errors: list[BaseException] = [] + + def first(): + try: + with sdk: + second_entered.wait(timeout=5) + except BaseException as exc: # surfaced in the assertion below + errors.append(exc) + first_exited.set() + + def second(): + try: + with sdk: + second_entered.set() + first_exited.wait(timeout=5) + # The other scope is gone; ours is still open. + sdk.teams.list() + except BaseException as exc: # surfaced in the assertion below + errors.append(exc) + + threads = [threading.Thread(target=first), threading.Thread(target=second)] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=5) + + assert not errors, f"a scope was torn down early: {errors}" + + def test_request_without_open_raises_sdk_error(self, api, sdk): + api.get("/teams").respond(200, json=[]) + + with pytest.raises(CodesphereError) as exc_info: + sdk.teams.list() + assert isinstance(exc_info.value, RuntimeError) + + +class TestFlagsCache: + def test_concurrent_first_reads_fetch_once(self, api, sdk): + route = api.get(FLAGS_ENDPOINT).respond(200, json=FLAGS_JSON) + + with sdk: + errors = _run_in_threads(sdk.flags.get, 8) + + assert not errors + assert route.call_count == 1 + + def test_invalidate_during_inflight_fetch_is_not_lost(self, api, sdk): + fetch_started = threading.Event() + release = threading.Event() + calls = 0 + + def flags(request): + nonlocal calls + calls += 1 + fetch_started.set() + release.wait(timeout=5) + return httpx.Response(200, json=FLAGS_JSON) + + api.get(FLAGS_ENDPOINT).mock(side_effect=flags) + + with sdk: + store = sdk._http_client.flags_store + worker = threading.Thread(target=sdk.flags.get) + worker.start() + fetch_started.wait(timeout=5) + + store.invalidate() + release.set() + worker.join(timeout=5) + + assert store.cached() is None, "invalidate() was silently discarded" + sdk.flags.get() + assert calls == 2, "the invalidated snapshot was served from cache" diff --git a/tests/test_concurrency.py b/tests/test_concurrency.py new file mode 100644 index 0000000..fff4d59 --- /dev/null +++ b/tests/test_concurrency.py @@ -0,0 +1,393 @@ +"""Concurrency regression tests for the async client. + +Each test encodes an invariant that one SDK instance shared across tasks must +uphold. Interleaving is driven by ``asyncio.Event`` handshakes rather than +wall-clock sleeps, so the ordering under test is deterministic and these do +not flake on a loaded machine. + +Tests assert the *invariant*, not the mechanism: they check that a failure +surfaces as a ``CodesphereError``, not that a particular exception class was +raised. That keeps them honest about what callers actually depend on. +""" + +import asyncio +import json +import time +from unittest.mock import AsyncMock, patch + +import httpx +import pytest +import respx + +from codesphere import ( + ClientStateError, + CodesphereSDK, + ConflictError, + NotFoundError, + RetryConfig, +) +from codesphere.exceptions import CodesphereError, StaleModelError +from codesphere.feature_flags import FLAGS_ENDPOINT + +BASE = "https://concurrency.test/api" + + +@pytest.fixture +def api(): + with respx.mock(base_url=BASE, assert_all_called=False) as router: + yield router + + +@pytest.fixture +def sdk(): + return CodesphereSDK(token="t", base_url=BASE, retry=RetryConfig(max_retries=0)) + + +@pytest.fixture +def sleep_mock(): + """Collapse retry backoff so retry tests stay fast.""" + with patch("asyncio.sleep", new_callable=AsyncMock) as mock: + yield mock + + +class TestClientLifecycle: + async def test_concurrent_scopes_do_not_close_each_other(self, api, sdk): + """Two tasks sharing one SDK: the first to exit must not tear down + the transport out from under the second.""" + api.get("/teams").respond(200, json=[]) + + first_exited = asyncio.Event() + second_entered = asyncio.Event() + + async def first(): + async with sdk: + # Hold the scope open until the second one is inside it too, + # so the two genuinely overlap. + await second_entered.wait() + first_exited.set() + + async def second(): + async with sdk: + second_entered.set() + await first_exited.wait() + # The other scope is gone; ours is still open. + return await sdk.teams.list() + + _, teams = await asyncio.gather(first(), second()) + assert teams == [] + + async def test_nested_scope_stays_open_until_outermost_exit(self, api, sdk): + api.get("/teams").respond(200, json=[]) + + async with sdk: + async with sdk: + await sdk.teams.list() + # Inner scope exited; the outer scope still owns an open client. + assert await sdk.teams.list() == [] + + async def test_transport_is_closed_once_every_scope_exits(self, api, sdk): + api.get("/teams").respond(200, json=[]) + + # Nesting is the point of this test, so it is not collapsible. + async with sdk: # noqa: SIM117 + async with sdk: + pass + assert sdk._http_client._client is None + + async def test_request_without_open_raises_sdk_error(self, api, sdk): + api.get("/teams").respond(200, json=[]) + + with pytest.raises(CodesphereError) as exc_info: + await sdk.teams.list() + # Kept as a RuntimeError too, so existing handlers keep working. + assert isinstance(exc_info.value, RuntimeError) + + async def test_close_during_retry_backoff_raises_sdk_error(self, api): + """Closing mid-backoff must not leak httpx's own RuntimeError.""" + api.get("/teams").respond(503, json={}) + sdk = CodesphereSDK( + token="t", + base_url=BASE, + retry=RetryConfig(max_retries=3, backoff_factor=0.3), + ) + await sdk.open() + + task = asyncio.create_task(sdk.teams.list()) + await asyncio.sleep(0.05) # let the first attempt fail into backoff + await sdk.close() + + with pytest.raises(CodesphereError): + await task + + +class TestFlagsCache: + @staticmethod + def _snapshot() -> dict: + return {"features": {"available": ["gadgets"], "enabled": ["gadgets"]}} + + async def test_invalidate_during_inflight_fetch_is_not_lost(self, api, sdk): + """A fetch that started before invalidate() must not repopulate the + cache it was told to drop.""" + fetch_started = asyncio.Event() + release = asyncio.Event() + calls = 0 + + async def flags(request): + nonlocal calls + calls += 1 + fetch_started.set() + await release.wait() + return httpx.Response(200, json=self._snapshot()) + + api.get(FLAGS_ENDPOINT).mock(side_effect=flags) + + async with sdk: + store = sdk._http_client.flags_store + task = asyncio.create_task(sdk.flags.get()) + await fetch_started.wait() + + store.invalidate() + release.set() + await task + + assert store.cached() is None, "invalidate() was silently discarded" + await sdk.flags.get() + assert calls == 2, "the invalidated snapshot was served from cache" + + async def test_concurrent_first_reads_fetch_once(self, api, sdk): + """Regression guard: the lock already dedupes: keep it that way.""" + route = api.get(FLAGS_ENDPOINT).respond(200, json=self._snapshot()) + + async with sdk: + await asyncio.gather(*(sdk.flags.get() for _ in range(8))) + + assert route.call_count == 1 + + +class TestRetrySideEffects: + async def test_retried_delete_404_is_treated_as_success( + self, api, sleep_mock, sample_workspace_data + ): + """The first DELETE reached the server behind a 503; the retry then + sees 404. The resource is gone, which is what the caller asked for.""" + api.get("/workspaces/72678").respond(200, json=sample_workspace_data) + route = api.delete("/workspaces/72678/landscape/teardown") + route.side_effect = [ + httpx.Response(503), + httpx.Response(404, json={"message": "no landscape deployed"}), + ] + + sdk = CodesphereSDK(token="t", base_url=BASE, retry=RetryConfig(max_retries=2)) + async with sdk: + workspace = await sdk.workspaces.get(72678) + await workspace.landscape.teardown() + + assert route.call_count == 2 + + async def test_first_attempt_404_still_raises( + self, api, sleep_mock, sample_workspace_data + ): + api.get("/workspaces/72678").respond(200, json=sample_workspace_data) + api.delete("/workspaces/72678/landscape/teardown").respond( + 404, json={"message": "no landscape deployed"} + ) + + sdk = CodesphereSDK(token="t", base_url=BASE, retry=RetryConfig(max_retries=2)) + async with sdk: + workspace = await sdk.workspaces.get(72678) + with pytest.raises(NotFoundError): + await workspace.landscape.teardown() + + async def test_retried_non_delete_404_still_raises(self, api, sleep_mock): + """The 404-as-success rule is scoped to DELETE. A PUT that 404s + after a retry is a genuine miss, not a completed write.""" + route = api.put("/workspaces/72678/env-vars") + route.side_effect = [ + httpx.Response(503), + httpx.Response(404, json={"message": "no such workspace"}), + ] + + sdk = CodesphereSDK(token="t", base_url=BASE, retry=RetryConfig(max_retries=2)) + async with sdk: + with pytest.raises(NotFoundError): + await sdk._http_client.request("PUT", "/workspaces/72678/env-vars") + + assert route.call_count == 2 + + +class TestModelStaleness: + async def test_concurrent_updates_never_leave_a_lying_model( + self, api, sdk, sample_workspace_data + ): + """The server orders writes by arrival, the client by response + latency. When those disagree, the local model must refuse to answer + rather than report a value the platform does not hold. + """ + from codesphere.models.workspace import WorkspaceUpdate + + api.get("/workspaces/72678").respond(200, json=sample_workspace_data) + + applied: list[str] = [] + second_applied = asyncio.Event() + + async def patch_workspace(request): + name = json.loads(request.content)["name"] + if name == "second": + # "second" reaches the server last, so the server ends on it. + await asyncio.sleep(0) + applied.append(name) + second_applied.set() + return httpx.Response(200) + applied.append(name) + # ...but its response comes back first, so a naive write-back + # would leave the model claiming "first". + await second_applied.wait() + return httpx.Response(200) + + api.patch("/workspaces/72678").mock(side_effect=patch_workspace) + + async with sdk: + workspace = await sdk.workspaces.get(72678) + await asyncio.gather( + workspace.update(WorkspaceUpdate(name="first")), + workspace.update(WorkspaceUpdate(name="second")), + ) + + server_value = applied[-1] + try: + local_value = workspace.name + except CodesphereError: + return # refused to answer: correct + assert local_value == server_value, ( + f"model reports {local_value!r} but the server holds " + f"{server_value!r} (apply order: {applied})" + ) + + async def test_refresh_restores_a_stale_model( + self, api, sdk, sample_workspace_data + ): + """refresh() is the documented way back: it re-reads the server and + the instance becomes readable again.""" + from codesphere.models.workspace import WorkspaceUpdate + + api.get("/workspaces/72678").respond(200, json=sample_workspace_data) + api.patch("/workspaces/72678").respond(200) + + async with sdk: + workspace = await sdk.workspaces.get(72678) + await workspace.update(WorkspaceUpdate(name="renamed")) + + with pytest.raises(StaleModelError): + _ = workspace.name + + # The server is the authority, not the payload we sent. + api.get("/workspaces/72678").respond( + 200, json={**sample_workspace_data, "name": "renamed-by-someone-else"} + ) + await workspace.refresh() + + assert workspace.name == "renamed-by-someone-else" + assert workspace.to_dict()["name"] == "renamed-by-someone-else" + + +class TestExternalActors: + """The platform can be changed by someone other than this client.""" + + @staticmethod + def _status(state: str, started_at: str | None) -> list[dict]: + return [ + { + "state": state, + "startedAt": started_at, + "finishedAt": None, + "steps": [{"state": state}], + "replica": "0", + "server": "web", + } + ] + + async def test_wait_for_stage_detects_a_foreign_restart( + self, api, sdk, sample_workspace_data + ): + """Someone else redeploying mid-wait must not be reported as our + run succeeding.""" + api.get("/workspaces/72678").respond(200, json=sample_workspace_data) + route = api.get("/workspaces/72678/pipeline/run") + route.side_effect = [ + httpx.Response(200, json=self._status("running", "2026-08-06T10:00:00Z")), + # Same server, different start time: this is a different run. + httpx.Response(200, json=self._status("success", "2026-08-06T10:05:00Z")), + ] + + async with sdk: + workspace = await sdk.workspaces.get(72678) + with pytest.raises(ConflictError, match="restarted while waiting"): + await workspace.landscape.wait_for_stage( + "run", timeout=30.0, poll_interval=0.01 + ) + + async def test_wait_for_stage_accepts_the_run_it_started( + self, api, sdk, sample_workspace_data + ): + api.get("/workspaces/72678").respond(200, json=sample_workspace_data) + route = api.get("/workspaces/72678/pipeline/run") + route.side_effect = [ + httpx.Response(200, json=self._status("running", "2026-08-06T10:00:00Z")), + httpx.Response(200, json=self._status("success", "2026-08-06T10:00:00Z")), + ] + + async with sdk: + workspace = await sdk.workspaces.get(72678) + result = await workspace.landscape.wait_for_stage( + "run", timeout=30.0, poll_interval=0.01 + ) + + assert len(result) == 1 + + async def test_wait_timeout_counts_request_time( + self, api, sdk, sample_workspace_data + ): + """A slow status endpoint must consume the timeout budget, not just + the sleeps between polls.""" + api.get("/workspaces/72678").respond(200, json=sample_workspace_data) + + async def slow_status(request): + await asyncio.sleep(0.15) + return httpx.Response(200, json={"isRunning": False}) + + api.get("/workspaces/72678/status").mock(side_effect=slow_status) + + async with sdk: + workspace = await sdk.workspaces.get(72678) + started = time.monotonic() + with pytest.raises(TimeoutError): + await workspace.wait_until_running(timeout=0.2, poll_interval=5.0) + elapsed = time.monotonic() - started + + # Old behavior slept a full poll_interval past the deadline. + assert elapsed < 1.0, ( + f"overran the deadline by a poll interval ({elapsed:.2f}s)" + ) + + async def test_log_stream_is_single_use(self, api, sdk, sample_workspace_data): + from codesphere.resources.workspace.logs import LogStage, StageTarget + + api.get("/workspaces/72678").respond(200, json=sample_workspace_data) + api.get("/workspaces/72678/logs/run/1").respond( + 200, + text='event: data\ndata: {"message": "hello"}\n\n', + headers={"content-type": "text/event-stream"}, + ) + + async with sdk: + workspace = await sdk.workspaces.get(72678) + stream = workspace.logs.open(StageTarget(stage=LogStage.RUN, step=1)) + + async with stream: + assert [entry.message async for entry in stream] == ["hello"] + # A second consumer would interleave frame parsing. + with pytest.raises(ClientStateError, match="already being consumed"): + stream.__aiter__() + + with pytest.raises(ClientStateError, match="already been opened"): + await stream.__aenter__() diff --git a/tests/test_http_client.py b/tests/test_http_client.py index d9cf081..f3ed92a 100644 --- a/tests/test_http_client.py +++ b/tests/test_http_client.py @@ -35,17 +35,32 @@ async def test_request_before_open_raises(self, transport): with pytest.raises(RuntimeError, match="not open"): await transport.request("GET", "/x") - async def test_open_is_idempotent(self, transport, api): + async def test_repeated_open_reuses_one_transport(self, transport, api): await transport.open() first = transport._client await transport.open() assert transport._client is first + + async def test_scopes_are_reference_counted(self, transport, api): + await transport.open() + await transport.open() + + await transport.close() + assert transport._client is not None, "an unbalanced close tore it down" + await transport.close() assert transport._client is None async def test_close_without_open_is_noop(self, transport): await transport.close() + async def test_surplus_close_does_not_underflow_the_count(self, transport, api): + await transport.close() + await transport.open() + assert transport._client is not None + await transport.close() + assert transport._client is None + class TestErrorMapping: @pytest.mark.parametrize(