From 24d2e42308807bfaa2ffa19276c57d848660453e Mon Sep 17 00:00:00 2001 From: Datata1 Date: Thu, 6 Aug 2026 16:15:09 +0200 Subject: [PATCH 1/7] test(concurrency): regression suite for six verified races Adds deterministic concurrency tests for the async client and its generated sync twin. Interleaving is driven by Event/Barrier handshakes rather than wall-clock sleeps, so the ordering under test is fixed. Tests assert the invariant (a failure surfaces as CodesphereError), not the exception class, so they encode what callers depend on and survive the fixes that follow. Eleven of these fail against the current client, each for a distinct defect: unrefcounted close(), a captured client across retry backoff, cross-thread TOCTOU in the sync open(), a lost flags invalidation, a retried DELETE whose 404 masks success, and in-place model mutation that diverges from server write ordering. --- tests/sync/test_concurrency.py | 168 +++++++++++++++++++++++ tests/test_concurrency.py | 241 +++++++++++++++++++++++++++++++++ 2 files changed, 409 insertions(+) create mode 100644 tests/sync/test_concurrency.py create mode 100644 tests/test_concurrency.py 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..42925c3 --- /dev/null +++ b/tests/test_concurrency.py @@ -0,0 +1,241 @@ +"""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 +from unittest.mock import AsyncMock, patch + +import httpx +import pytest +import respx + +from codesphere import CodesphereSDK, NotFoundError, RetryConfig +from codesphere.exceptions import CodesphereError +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() + + +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})" + ) From 32a81ad50fa886e8abd06bc1c1c9edbe9ab9cca5 Mon Sep 17 00:00:00 2001 From: Datata1 Date: Thu, 6 Aug 2026 16:15:24 +0200 Subject: [PATCH 2/7] fix(client): reference-count transport scopes, map closed-client errors Sharing one SDK across tasks was unsafe: open() was idempotent but close() was not, so the first scope to exit tore the transport down for everyone else. open()/close() are now reference counted behind _compat.Lock, which also closes the check-then-act window that let two threads in the sync twin each build a transport and orphan one. The teardown awaits __aexit__ outside the lock so a slow drain cannot block a concurrent open() of the next transport. request() now re-resolves the client on every retry attempt instead of capturing it once, so a close during backoff is caught by the SDK rather than surfacing as httpx's bare RuntimeError. That error is remapped to the new ClientStateError(CodesphereError, RuntimeError) only when the client is actually gone; unrelated httpx internals still propagate. The dual base keeps existing `except RuntimeError` handlers working. test_open_is_idempotent asserted the old (buggy) teardown semantics and is replaced by tests for reuse, balanced counting, and close underflow. --- src/codesphere/__init__.py | 2 + src/codesphere/_async/http_client.py | 63 +++++++++++++++++++++------- src/codesphere/_sync/http_client.py | 63 +++++++++++++++++++++------- src/codesphere/exceptions.py | 18 ++++++++ tests/test_http_client.py | 17 +++++++- 5 files changed, 134 insertions(+), 29 deletions(-) diff --git a/src/codesphere/__init__.py b/src/codesphere/__init__.py index 5b3aa2a..bbb41bb 100644 --- a/src/codesphere/__init__.py +++ b/src/codesphere/__init__.py @@ -33,6 +33,7 @@ APIError, AuthenticationError, AuthorizationError, + ClientStateError, CodesphereError, ConflictError, FeatureFlagError, @@ -73,6 +74,7 @@ "AuthorizationError", "CategoryFlags", "Characteristic", + "ClientStateError", "CodesphereError", "CodesphereSDK", "ConflictError", diff --git a/src/codesphere/_async/http_client.py b/src/codesphere/_async/http_client.py index 554118b..12df816 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__) @@ -37,6 +43,10 @@ class APIHttpClient: 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 +61,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 +79,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 +101,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 +115,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 +142,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 +159,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) diff --git a/src/codesphere/_sync/http_client.py b/src/codesphere/_sync/http_client.py index 46edbb9..0c3451d 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 @@ -40,6 +46,10 @@ class APIHttpClient: 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 +64,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 +82,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 +104,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 +118,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 +143,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 +160,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) diff --git a/src/codesphere/exceptions.py b/src/codesphere/exceptions.py index a3d520e..45b2e81 100644 --- a/src/codesphere/exceptions.py +++ b/src/codesphere/exceptions.py @@ -160,6 +160,24 @@ def __init__(self, message: str | None = None): super().__init__(message) +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/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( From 5bed327a56079bb8198d769d8c240e793164091c Mon Sep 17 00:00:00 2001 From: Datata1 Date: Thu, 6 Aug 2026 16:17:17 +0200 Subject: [PATCH 3/7] fix(flags): honor invalidate() against an in-flight fetch invalidate() cleared the snapshot outside the lock, so a get() already awaiting its request would overwrite the None with data fetched before the invalidation. The invalidation evaporated and the next read served the stale snapshot. A fetch now stamps the generation it started under and installs its result only if invalidate() has not bumped it meanwhile. invalidate() stays synchronous: taking the lock there would block the caller for the duration of an in-flight flags request. require() also read _legacy_platform outside the lock after get() returned, so a concurrent refresh could pair a snapshot with the legacy flag of a different fetch. _fetch() now returns both, and the new _get_with_legacy() hands them to callers from one lock acquisition. --- src/codesphere/_async/flags_store.py | 51 ++++++++++++++++++++++------ src/codesphere/_sync/flags_store.py | 49 ++++++++++++++++++++------ 2 files changed, 78 insertions(+), 22 deletions(-) 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/_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 From eb0de6a21a56ba91e5259fbb4f452dc8a7698cef Mon Sep 17 00:00:00 2001 From: Datata1 Date: Thu, 6 Aug 2026 16:19:56 +0200 Subject: [PATCH 4/7] fix(retry): treat a 404 on a retried DELETE as success When a DELETE reached the platform and only its response was lost (a gateway 503), the retry saw 404 and the SDK raised NotFoundError for an operation that had in fact succeeded. Tearing down a landscape reported "no landscape deployed" precisely because the teardown worked. A 404 is now treated as success when the request is a DELETE and at least one retry has already run. The attempt > 0 guard matters: a 404 on the first attempt is a genuine miss and still raises. The rule is scoped to DELETE, so a retried PUT that 404s is unaffected. Safe for every current DELETE operation: all five declare response_model=NoneType, so the response body is never parsed. The RetryConfig docstring now states the residual risk this does not solve. PUT and DELETE are idempotent as methods but not always in effect, and without idempotency keys the SDK cannot stop a retried teardown from executing twice. Callers who cannot tolerate that are pointed at max_retries=0. --- src/codesphere/_async/http_client.py | 19 +++++++++++++++++++ src/codesphere/_sync/http_client.py | 19 +++++++++++++++++++ src/codesphere/config.py | 17 +++++++++++++++++ tests/test_concurrency.py | 16 ++++++++++++++++ 4 files changed, 71 insertions(+) diff --git a/src/codesphere/_async/http_client.py b/src/codesphere/_async/http_client.py index 12df816..7ff8394 100644 --- a/src/codesphere/_async/http_client.py +++ b/src/codesphere/_async/http_client.py @@ -37,6 +37,15 @@ 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. @@ -201,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/_sync/http_client.py b/src/codesphere/_sync/http_client.py index 0c3451d..691b972 100644 --- a/src/codesphere/_sync/http_client.py +++ b/src/codesphere/_sync/http_client.py @@ -40,6 +40,15 @@ 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. @@ -202,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/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/tests/test_concurrency.py b/tests/test_concurrency.py index 42925c3..568b196 100644 --- a/tests/test_concurrency.py +++ b/tests/test_concurrency.py @@ -191,6 +191,22 @@ async def test_first_attempt_404_still_raises( 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( From c5ee4391929509dc96ffeb7fb8e69f03f6598eb1 Mon Sep 17 00:00:00 2001 From: Datata1 Date: Thu, 6 Aug 2026 16:37:43 +0200 Subject: [PATCH 5/7] feat(models)!: invalidate models after a write instead of guessing BREAKING CHANGE: reading a field on a model after update() raises StaleModelError until refresh() is called. Writes copied values back into the local model, which is wrong by construction under concurrency: the platform orders writes by arrival while the client sees them by response. Two tasks updating the same workspace could leave the model reporting "first" while the server held "second", permanently and silently. Workspace.update() wrote back the request payload; the Domain methods wrote back the server's response, which is authoritative for the moment it was produced but can still be overtaken by a concurrent write. Both are now handled the same way: the instance is marked stale. Staleness drops the field values from __dict__ so reads route through __getattr__, which costs nothing until a model actually goes stale. Identity fields survive, so a stale instance can still be logged and re-fetched, and repr() marks it. model_dump/model_dump_json are guarded too: pydantic serializes straight from __dict__, so without that a stale model would quietly dump only its identity. Workspace.refresh() and Domain.refresh() re-read the entity in place. Domain writes still return the server's response; prefer it over self. Removes codesphere.utils.update_model_fields, now unused. --- CHANGELOG.md | 30 ++++++++ src/codesphere/__init__.py | 2 + src/codesphere/_async/core/base.py | 72 ++++++++++++++++++- .../_async/resources/team/domain/schemas.py | 30 ++++++-- .../_async/resources/workspace/schemas.py | 22 +++++- src/codesphere/_sync/core/base.py | 72 ++++++++++++++++++- .../_sync/resources/team/domain/schemas.py | 30 ++++++-- .../_sync/resources/workspace/schemas.py | 20 +++++- src/codesphere/exceptions.py | 21 ++++++ src/codesphere/utils.py | 16 ----- tests/resources/workspace/test_workspace.py | 22 +++++- tests/test_concurrency.py | 28 +++++++- 12 files changed, 331 insertions(+), 34 deletions(-) delete mode 100644 src/codesphere/utils.py diff --git a/CHANGELOG.md b/CHANGELOG.md index d835e16..f3b0fc4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,8 +14,36 @@ quickstart, guides, full API reference, and `llms.txt`; deployed to GitHub Pages on pushes to `main`. +### Fixed + +- 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 +60,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/src/codesphere/__init__.py b/src/codesphere/__init__.py index bbb41bb..2b7d456 100644 --- a/src/codesphere/__init__.py +++ b/src/codesphere/__init__.py @@ -42,6 +42,7 @@ NetworkError, NotFoundError, RateLimitError, + StaleModelError, TimeoutError, ValidationError, ) @@ -95,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/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/schemas.py b/src/codesphere/_async/resources/workspace/schemas.py index 27e8f15..cbb43b8 100644 --- a/src/codesphere/_async/resources/workspace/schemas.py +++ b/src/codesphere/_async/resources/workspace/schemas.py @@ -14,7 +14,6 @@ WorkspaceStatus, WorkspaceUpdate, ) -from codesphere.utils import update_model_fields from ...core.base import BoundModel from .env_vars import EnvVar, WorkspaceEnvVarManager @@ -43,10 +42,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 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/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/schemas.py b/src/codesphere/_sync/resources/workspace/schemas.py index 6a5089b..635519f 100644 --- a/src/codesphere/_sync/resources/workspace/schemas.py +++ b/src/codesphere/_sync/resources/workspace/schemas.py @@ -16,7 +16,6 @@ WorkspaceStatus, WorkspaceUpdate, ) -from codesphere.utils import update_model_fields from ...core.base import BoundModel from .env_vars import EnvVar, WorkspaceEnvVarManager @@ -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 diff --git a/src/codesphere/exceptions.py b/src/codesphere/exceptions.py index 45b2e81..672ea9c 100644 --- a/src/codesphere/exceptions.py +++ b/src/codesphere/exceptions.py @@ -160,6 +160,27 @@ 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. 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/test_concurrency.py b/tests/test_concurrency.py index 568b196..e91e309 100644 --- a/tests/test_concurrency.py +++ b/tests/test_concurrency.py @@ -19,7 +19,7 @@ import respx from codesphere import CodesphereSDK, NotFoundError, RetryConfig -from codesphere.exceptions import CodesphereError +from codesphere.exceptions import CodesphereError, StaleModelError from codesphere.feature_flags import FLAGS_ENDPOINT BASE = "https://concurrency.test/api" @@ -255,3 +255,29 @@ async def patch_workspace(request): 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" From 0d391fc5777296818dc1ba8305351db0c157c0ee Mon Sep 17 00:00:00 2001 From: Datata1 Date: Fri, 7 Aug 2026 16:43:54 +0200 Subject: [PATCH 6/7] fix(landscape): detect foreign restarts and stop overrunning deadlines wait_for_stage() could report success for a pipeline run the caller never started: if someone else redeployed mid-wait, it simply observed whatever run was current. The API exposes no run id, but started_at changes on restart, so the run is pinned on first sighting and a different value now raises ConflictError. Both wait_for_stage() and wait_until_running() counted elapsed time by summing poll_interval, ignoring how long the status requests took. A slow endpoint could overrun the timeout by a wide margin (5.15s against a 0.2s budget in the regression test). Both now measure against a monotonic deadline and never sleep past it. LogStream is backed by one SSE response body that can only be read once. Re-entering it stranded the first stream context, and a second iterator surfaced a raw httpx.StreamConsumed. Both now raise ClientStateError explaining that streams are single-use. Adds docs/guides/concurrency.md. Alongside the guarantees, it states plainly what the SDK cannot do: with no ETags, If-Match, or idempotency keys, read-modify-write cycles have an unclosable lost-update window and retried writes can execute twice. --- CHANGELOG.md | 17 ++ docs/guides/concurrency.md | 162 ++++++++++++++++++ docs/llms.txt | 1 + docs/reference/config.md | 2 + mkdocs.yml | 1 + .../workspace/landscape/resources.py | 103 +++++++---- .../resources/workspace/logs/resources.py | 31 +++- .../_async/resources/workspace/schemas.py | 25 +-- .../workspace/landscape/resources.py | 104 +++++++---- .../resources/workspace/logs/resources.py | 31 +++- .../_sync/resources/workspace/schemas.py | 26 +-- tests/test_concurrency.py | 112 +++++++++++- 12 files changed, 530 insertions(+), 85 deletions(-) create mode 100644 docs/guides/concurrency.md diff --git a/CHANGELOG.md b/CHANGELOG.md index f3b0fc4..46477c5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,8 +14,25 @@ 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 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/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 cbb43b8..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 @@ -87,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/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 635519f..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 @@ -22,7 +23,6 @@ from .git import WorkspaceGitManager from .landscape import WorkspaceLandscapeManager from .logs import WorkspaceLogManager -import time log = logging.getLogger(__name__) @@ -88,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/tests/test_concurrency.py b/tests/test_concurrency.py index e91e309..fff4d59 100644 --- a/tests/test_concurrency.py +++ b/tests/test_concurrency.py @@ -12,13 +12,20 @@ import asyncio import json +import time from unittest.mock import AsyncMock, patch import httpx import pytest import respx -from codesphere import CodesphereSDK, NotFoundError, RetryConfig +from codesphere import ( + ClientStateError, + CodesphereSDK, + ConflictError, + NotFoundError, + RetryConfig, +) from codesphere.exceptions import CodesphereError, StaleModelError from codesphere.feature_flags import FLAGS_ENDPOINT @@ -281,3 +288,106 @@ async def test_refresh_restores_a_stale_model( 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__() From 424b98ef88770caff257c1b96cd7a8cc3076c6ca Mon Sep 17 00:00:00 2001 From: Datata1 Date: Fri, 7 Aug 2026 16:48:23 +0200 Subject: [PATCH 7/7] build(sync-gen): do not abort generation when unasyncd rewrites files MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit unasyncd exits non-zero whenever it transforms a file, the way a formatter signals "changed". Since run() defaulted to fatal, the very run that did the work aborted before restore_all_blocks() and before ruff normalized the generated tree — so `make sync-gen` failed on every real change and had to be run twice to produce a correct result. --- scripts/gen_sync.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) 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;