From 79b74eca4eb0e79e6b8a242e3b818c17608cdc35 Mon Sep 17 00:00:00 2001 From: Ignacio Van Droogenbroeck <64545348+xe-nvdk@users.noreply.github.com> Date: Fri, 2 Oct 2026 11:57:01 -0600 Subject: [PATCH 1/3] fix(http): raise a distinct exception per Arc refusal on the database listings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Arc's three database listing endpoints carried no authentication at all. They now require read permission and, where the server restricts reads per database, a grant covering the database being listed. handle_response_error mapped both 401 and 403 onto ArcAuthenticationError, so a caller could not tell "re-authenticate" from "this token lacks a permission" — and 403 is not one condition either. Arc answers it for a token with no read permission, for a token whose grants do not cover the database asked for, for a licence-gated feature, and for its own permission data being unreadable, distinguishing them only by message. ArcAuthenticationError now means 401 only. Added: - ArcPermissionError: the token does not carry the permission the endpoint requires. - ArcScopedAccessError: Arc's per-database denial. The token may read, but not what was asked for. Carries the refused database on .database — empty or "*" when the call asked for every database, since Arc refuses databases.list() for a tenant-scoped token rather than returning a filtered list, matching SHOW DATABASES. A normal state, not a fault to retry. - ArcPermissionDataUnavailableError: Arc could not load the token's grants and failed closed. A server-side fault that says nothing about the token, so it is checked before the scoped shape and never reported as one. All subclass ArcPermissionError, which subclasses ArcAuthenticationError, so existing except blocks that caught 403 as an authentication failure keep working. Matching is by stable message fragment and covers both the read middleware's wording and the handler gates', so a change to the permission word cannot reclassify one refusal as another; an unrecognised 403 falls back to ArcPermissionError. Docstrings on list/get/list_measurements, sync and async, document which exception each refusal raises. --- CHANGELOG.md | 16 + README.md | 29 +- src/arc_client/__init__.py | 6 + src/arc_client/exceptions.py | 60 ++++ src/arc_client/http/base.py | 113 ++++++- src/arc_client/management/async_databases.py | 40 ++- src/arc_client/management/databases.py | 49 ++- tests/unit/http/test_databases_permissions.py | 280 ++++++++++++++++++ 8 files changed, 584 insertions(+), 9 deletions(-) create mode 100644 tests/unit/http/test_databases_permissions.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 1eb68f4..447d024 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -33,6 +33,15 @@ plus bug fixes and the first real HTTP-layer test coverage. `retry_backoff`. - New exceptions: `ArcNotSupportedError` (501) and `ArcLicenseError` (403 for a missing enterprise-license feature; subclass of `ArcAuthenticationError`). +- `ArcPermissionError` (403: the token lacks the permission the endpoint + requires), `ArcScopedAccessError` (403 from Arc's per-database check: the + token may read, but not the database asked for) and + `ArcPermissionDataUnavailableError` (403 because Arc could not load the + token's grants and failed closed — a server-side fault, not a statement + about the token). `ArcScopedAccessError` carries the refused database on + `.database`, empty or `"*"` when the call asked for every database. The + latter two subclass `ArcPermissionError`, which subclasses + `ArcAuthenticationError`, so existing `except` blocks keep working. - Real HTTP round-trip test coverage (via `pytest-httpx`) for headers, error mapping, compression on the wire, retries, msgpack decode, imports, databases, and enterprise wrappers — sync and async. Opt-in live integration tests under @@ -46,6 +55,13 @@ plus bug fixes and the first real HTTP-layer test coverage. from the payload's magic bytes. - `User-Agent` is now derived from the package version instead of a hardcoded string, and the package version is a single source of truth read by the build. +- HTTP 403 no longer collapses onto `ArcAuthenticationError`, which now means + only 401. The database listing endpoints need read permission and, where the + server restricts reads per database, a grant for the database — and Arc + refuses `databases.list()` for a tenant-scoped token rather than returning a + filtered list, matching `SHOW DATABASES`. That refusal is an expected state, + not a fault to retry: it raises `ArcScopedAccessError` naming the database, + while a token with no read permission raises `ArcPermissionError`. ### Fixed diff --git a/README.md b/README.md index 8a4b6da..961996e 100644 --- a/README.md +++ b/README.md @@ -192,6 +192,30 @@ measurements = client.databases.list_measurements("metrics") client.databases.delete("metrics", confirm=True) ``` +The three read methods need a token with read permission. On a server that +restricts reads per database they also need a grant for the database in +question, and `list()` needs one covering every database — Arc refuses rather +than returning a filtered list, the same way `SHOW DATABASES` does: + +```python +from arc_client.exceptions import ArcScopedAccessError + +try: + databases = client.databases.list().databases +except ArcScopedAccessError: + # Expected for a tenant-scoped token: it cannot enumerate every database, + # so ask for the one it is granted instead of retrying. + measurements = client.databases.list_measurements("metrics") +``` + +`ArcScopedAccessError.database` carries the database the call was refused for +(empty or `"*"` when the call asked for every database). A token with no read +permission at all raises `ArcPermissionError` instead; Arc failing to load the +token's grants raises `ArcPermissionDataUnavailableError`, a server-side fault +rather than a statement about the token; and a database that does not exist +still raises `ArcNotFoundError` — "you may not read it" and "it is not there" +stay apart. + ## Querying Data ### JSON Response @@ -445,7 +469,10 @@ the codec from the payload's magic bytes, so no `Content-Encoding` header is sen from arc_client.exceptions import ( ArcError, # Base exception ArcConnectionError, # Connection failures - ArcAuthenticationError,# Auth failures (401/403) + ArcAuthenticationError,# Token missing, invalid or expired (401) + ArcPermissionError, # Token lacks the required permission (403) + ArcScopedAccessError, # Token may read, but not this database (403) + ArcPermissionDataUnavailableError, # Arc could not load the grants (403) ArcLicenseError, # Enterprise feature not licensed (403) ArcQueryError, # Query execution errors ArcIngestionError, # Write failures diff --git a/src/arc_client/__init__.py b/src/arc_client/__init__.py index 11e22c3..fcbf45b 100644 --- a/src/arc_client/__init__.py +++ b/src/arc_client/__init__.py @@ -17,8 +17,11 @@ ArcLicenseError, ArcNotFoundError, ArcNotSupportedError, + ArcPermissionDataUnavailableError, + ArcPermissionError, ArcQueryError, ArcRateLimitError, + ArcScopedAccessError, ArcServerError, ArcValidationError, ) @@ -42,6 +45,9 @@ "ArcServerError", "ArcNotSupportedError", "ArcLicenseError", + "ArcPermissionError", + "ArcPermissionDataUnavailableError", + "ArcScopedAccessError", # Version "__version__", ] diff --git a/src/arc_client/exceptions.py b/src/arc_client/exceptions.py index a7b36bf..a0244ff 100644 --- a/src/arc_client/exceptions.py +++ b/src/arc_client/exceptions.py @@ -79,3 +79,63 @@ class ArcLicenseError(ArcAuthenticationError): """ pass + + +class ArcPermissionError(ArcAuthenticationError): + """HTTP 403: the token does not carry the permission the endpoint needs. + + Distinct from :class:`ArcAuthenticationError`, which now means only HTTP + 401 — the token is missing, invalid or expired. A 401 is fixed by + re-authenticating; a 403 is not, and needs a token with different + permissions. + + Subclasses :class:`ArcAuthenticationError` so existing ``except`` blocks + that caught 403 as an authentication failure keep working. + """ + + pass + + +class ArcPermissionDataUnavailableError(ArcPermissionError): + """HTTP 403 because Arc could not load the token's grants. + + Arc fails closed when its permission data cannot be read, which arrives + as a 403 like a real refusal. Nothing about the token or the database + caused it, so it is raised separately: retrying with a different database + or a different token will not help, and the Arc server log is where the + cause is. + """ + + pass + + +class ArcScopedAccessError(ArcPermissionError): + """HTTP 403 from Arc's per-database permission check. + + The token may read, but not the database the call asked about. Raised by + the database listing endpoints: + + * :meth:`arc_client.management.databases.DatabaseClient.list` — Arc + refuses to list every database for a token restricted to particular + ones rather than returning a filtered list, matching ``SHOW + DATABASES``. ``database`` is then ``"*"``. + * :meth:`~arc_client.management.databases.DatabaseClient.get` and + :meth:`~arc_client.management.databases.DatabaseClient.list_measurements` + — the token holds no grant for the database named, which ``database`` + carries. + + This is a normal state for a tenant-scoped token, not a transient + failure: an identical retry is refused identically, and the way forward + is to name a database the token is granted. It is deliberately separate + from :class:`ArcPermissionError` so "you are scoped, name your database" + and "this token cannot read at all" can be handled differently. + + Attributes: + database: The database the call was refused for, or ``"*"`` when it + asked for every database. ``None`` if Arc's message did not name + one. + """ + + def __init__(self, message: str, database: Optional[str] = None) -> None: + super().__init__(message) + self.database = database diff --git a/src/arc_client/http/base.py b/src/arc_client/http/base.py index d92e359..ffa588b 100644 --- a/src/arc_client/http/base.py +++ b/src/arc_client/http/base.py @@ -3,6 +3,7 @@ from __future__ import annotations import os +import re from typing import Any, Optional import httpx @@ -15,7 +16,10 @@ ArcLicenseError, ArcNotFoundError, ArcNotSupportedError, + ArcPermissionDataUnavailableError, + ArcPermissionError, ArcRateLimitError, + ArcScopedAccessError, ArcServerError, ) @@ -38,8 +42,83 @@ def build_headers( return headers +# Arc sends HTTP 403 for three different things and distinguishes them only +# by message, so the message is what these read. The shapes its read +# middleware and handler gates produce: +# +# no permission for read on database 'x' -> scoped +# access denied: no read permission for database 'x' -> scoped +# permission data unavailable -> grants unreadable +# token does not have 'read' permission -> no read permission +# Permission denied: read required -> no read permission +# +# Only the first two are resolved by naming a different database, so they +# must not collapse onto one exception. The patterns match the stable parts +# of each message rather than the whole string, so a change to the permission +# word or the surrounding punctuation cannot silently reclassify one as +# another, and the database name is captured where Arc sends one. +_SCOPED_DENIAL_RES = ( + # The read middleware on the listing routes. The name is empty on the + # list-everything route, which sends no database. + re.compile( + r"^no permission for \w+ on database '(?P[^']*)'", + re.IGNORECASE, + ), + # The handler gates' own wording. + re.compile( + r"^access denied: no \w+ permission for database '(?P[^']*)'", + re.IGNORECASE, + ), +) + +# Arc could not load the token's grants and refused rather than guess. A 403 +# like the others, but it says nothing about the token or the database, so it +# must not be reported as a permission problem. Checked first. +_PERMISSION_DATA_UNAVAILABLE = "permission data unavailable" + + +def _scoped_denial_database(message: object) -> Optional[str]: + """Return the database named by a per-database denial, else None. + + An empty capture is returned as ``""`` — Arc sends an empty name on the + list-everything route — so callers must test against ``None``, not + truthiness, to tell "not a scoped denial" from "scoped, no name". + """ + if not isinstance(message, str): + return None + text = message.strip() + for pattern in _SCOPED_DENIAL_RES: + match = pattern.match(text) + if match is not None: + return match.group("database") + return None + + +def _is_permission_data_unavailable(message: object) -> bool: + """True if Arc failed closed because it could not load the grants.""" + return isinstance(message, str) and _PERMISSION_DATA_UNAVAILABLE in message.lower() + + def handle_response_error(response: httpx.Response) -> None: - """Handle HTTP error responses and raise appropriate exceptions.""" + """Handle HTTP error responses and raise appropriate exceptions. + + Raises: + ArcAuthenticationError: HTTP 401 — the token is missing, invalid or + expired. Re-authenticate. + ArcLicenseError: HTTP 403 naming an enterprise license. + ArcScopedAccessError: HTTP 403 from Arc's per-database permission + check — the token may read, but not the database asked for. A + subclass of ``ArcPermissionError``; carries ``database``. + ArcPermissionDataUnavailableError: HTTP 403 because Arc could not + load the token's grants and failed closed. A server-side fault, + not a statement about the token. + ArcPermissionError: any other HTTP 403 — the token does not carry the + permission the endpoint requires. + ArcNotFoundError: HTTP 404. + ArcRateLimitError: HTTP 429. + ArcNotSupportedError: HTTP 501. + ArcServerError: HTTP 5xx and any other non-2xx status. + """ if response.is_success: return @@ -58,7 +137,37 @@ def handle_response_error(response: httpx.Response) -> None: # ArcAuthenticationError, so plain permission-denied handling still works). if isinstance(message, str) and "enterprise license" in message.lower(): raise ArcLicenseError(f"Enterprise license required: {message}") - raise ArcAuthenticationError(f"Permission denied: {message}") + # Checked before the scoped shape: Arc failing to read its own + # permission data is a server fault, and reporting it as "your token + # is scoped" would send the caller after a database that was never + # the problem. + if _is_permission_data_unavailable(message): + raise ArcPermissionDataUnavailableError( + f"Arc could not read its own permission data and refused the request " + f"rather than guessing. This is a server-side fault, not a problem " + f"with this token — check the Arc server log. ({message})" + ) + # Arc's per-database check refuses a token that may read but holds no + # grant for the database asked for. Naming a database it is granted is + # the only thing that resolves it, so it gets its own exception rather + # than being reported as "this token cannot read". + database = _scoped_denial_database(message) + if database is not None: + if database in ("", "*"): + detail = ( + "this token is scoped to specific databases, so Arc will not list " + "them all. Name one with get() or list_measurements()." + ) + else: + detail = ( + f"this token is scoped to specific databases and holds no read " + f"grant for {database!r}. Name a database it is granted." + ) + raise ArcScopedAccessError( + f"Access denied: {detail} ({message})", + database=database, + ) + raise ArcPermissionError(f"Permission denied: {message}") elif status_code == 404: raise ArcNotFoundError(f"Resource not found: {message}") elif status_code == 429: diff --git a/src/arc_client/management/async_databases.py b/src/arc_client/management/async_databases.py index db61aa2..c3c6bb0 100644 --- a/src/arc_client/management/async_databases.py +++ b/src/arc_client/management/async_databases.py @@ -23,7 +23,21 @@ def __init__(self, http: AsyncHTTPClient, config: ClientConfig) -> None: self._config = config async def list(self) -> DatabaseListResponse: - """List all databases.""" + """List all databases. + + Needs a token with read permission. Where the server restricts reads + per database, it also needs a grant covering every database: Arc + refuses this call for a token scoped to particular databases rather + than returning a filtered list, the same way ``SHOW DATABASES`` does. + + Raises: + ArcScopedAccessError: The token is scoped to specific databases + (``database`` is ``"*"``). Expected for a tenant-scoped + token — name a database with :meth:`get` or + :meth:`list_measurements` instead of retrying. + ArcPermissionError: The token carries no read permission. + ArcAuthenticationError: The token is missing, invalid or expired. + """ try: response = await self._http.get("/api/v1/databases") return DatabaseListResponse.model_validate(response.json()) @@ -33,7 +47,18 @@ async def list(self) -> DatabaseListResponse: raise ArcError(f"Failed to list databases: {e}") from e async def get(self, name: str) -> DatabaseInfo: - """Get a single database by name.""" + """Get a single database by name. + + Raises: + ArcScopedAccessError: The token holds no read grant for ``name`` + (carried on ``database``). It may be able to read others. + ArcPermissionError: The token carries no read permission. + ArcAuthenticationError: The token is missing, invalid or expired. + ArcNotFoundError: No such database. Kept separate from the + permission errors: "you may not read it" and "it is not + there" are different answers. + ArcValidationError: The name is not a valid database name. + """ validate_name(name, "database") try: response = await self._http.get(f"/api/v1/databases/{name}") @@ -44,7 +69,16 @@ async def get(self, name: str) -> DatabaseInfo: raise ArcError(f"Failed to get database {name!r}: {e}") from e async def list_measurements(self, name: str) -> MeasurementListResponse: - """List measurements within a database.""" + """List measurements within a database. + + Raises: + ArcScopedAccessError: The token holds no read grant for ``name`` + (carried on ``database``). Ask for a database it is granted. + ArcPermissionError: The token carries no read permission. + ArcAuthenticationError: The token is missing, invalid or expired. + ArcNotFoundError: No such database. + ArcValidationError: The name is not a valid database name. + """ validate_name(name, "database") try: response = await self._http.get(f"/api/v1/databases/{name}/measurements") diff --git a/src/arc_client/management/databases.py b/src/arc_client/management/databases.py index b89ccb5..262b138 100644 --- a/src/arc_client/management/databases.py +++ b/src/arc_client/management/databases.py @@ -21,6 +21,15 @@ class DatabaseClient: Create and delete require an admin token. Delete additionally requires the server to have delete operations enabled and an explicit ``confirm=True``. + The three read methods need a token with read permission. On a server that + restricts reads per database they also need a grant for the database in + question, and :meth:`list` needs one covering every database — Arc refuses + rather than returning a filtered list, matching ``SHOW DATABASES``. That + refusal raises :class:`~arc_client.exceptions.ArcScopedAccessError`, which + is a normal state for a tenant-scoped token rather than a fault to retry; + a token with no read permission at all raises + :class:`~arc_client.exceptions.ArcPermissionError`. + Example: >>> client.databases.create("metrics") >>> for db in client.databases.list().databases: @@ -32,7 +41,21 @@ def __init__(self, http: SyncHTTPClient, config: ClientConfig) -> None: self._config = config def list(self) -> DatabaseListResponse: - """List all databases.""" + """List all databases. + + Needs a token with read permission. Where the server restricts reads + per database, it also needs a grant covering every database: Arc + refuses this call for a token scoped to particular databases rather + than returning a filtered list, the same way ``SHOW DATABASES`` does. + + Raises: + ArcScopedAccessError: The token is scoped to specific databases + (``database`` is ``"*"``). Expected for a tenant-scoped + token — name a database with :meth:`get` or + :meth:`list_measurements` instead of retrying. + ArcPermissionError: The token carries no read permission. + ArcAuthenticationError: The token is missing, invalid or expired. + """ try: response = self._http.get("/api/v1/databases") return DatabaseListResponse.model_validate(response.json()) @@ -42,7 +65,18 @@ def list(self) -> DatabaseListResponse: raise ArcError(f"Failed to list databases: {e}") from e def get(self, name: str) -> DatabaseInfo: - """Get a single database by name.""" + """Get a single database by name. + + Raises: + ArcScopedAccessError: The token holds no read grant for ``name`` + (carried on ``database``). It may be able to read others. + ArcPermissionError: The token carries no read permission. + ArcAuthenticationError: The token is missing, invalid or expired. + ArcNotFoundError: No such database. Kept separate from the + permission errors: "you may not read it" and "it is not + there" are different answers. + ArcValidationError: The name is not a valid database name. + """ validate_name(name, "database") try: response = self._http.get(f"/api/v1/databases/{name}") @@ -53,7 +87,16 @@ def get(self, name: str) -> DatabaseInfo: raise ArcError(f"Failed to get database {name!r}: {e}") from e def list_measurements(self, name: str) -> MeasurementListResponse: - """List measurements within a database.""" + """List measurements within a database. + + Raises: + ArcScopedAccessError: The token holds no read grant for ``name`` + (carried on ``database``). Ask for a database it is granted. + ArcPermissionError: The token carries no read permission. + ArcAuthenticationError: The token is missing, invalid or expired. + ArcNotFoundError: No such database. + ArcValidationError: The name is not a valid database name. + """ validate_name(name, "database") try: response = self._http.get(f"/api/v1/databases/{name}/measurements") diff --git a/tests/unit/http/test_databases_permissions.py b/tests/unit/http/test_databases_permissions.py new file mode 100644 index 0000000..e373437 --- /dev/null +++ b/tests/unit/http/test_databases_permissions.py @@ -0,0 +1,280 @@ +"""Arc's answers for the database listing endpoints. + +The three ``GET /api/v1/databases*`` routes require read permission and, on a +server that restricts reads per database, a grant for the database asked for. +A 401, the two different 403s and a 404 are four different answers and must +reach callers as four different exceptions. +""" + +from __future__ import annotations + +import pytest +from pytest_httpx import HTTPXMock + +from arc_client import ArcClient, AsyncArcClient +from arc_client.exceptions import ( + ArcAuthenticationError, + ArcError, + ArcNotFoundError, + ArcPermissionDataUnavailableError, + ArcPermissionError, + ArcScopedAccessError, +) + +# The bodies the server actually sends. Every 403 here shares a status code +# and differs only in message, so the message is the whole discriminator. +NO_TOKEN = {"success": False, "error": "Authentication required"} +BAD_TOKEN = {"success": False, "error": "Invalid or expired token"} +# Coarse refusals: the token carries no read permission at all. +NO_READ_PERMISSION = {"success": False, "error": "token does not have 'read' permission"} +OSS_NO_READ = {"success": False, "error": "Permission denied: read required"} +# The read middleware's RBAC denial, which is what these routes return in +# practice. The name is empty on the list-everything route, which sends none. +SCOPED_ALL = {"success": False, "error": "no permission for read on database ''"} +SCOPED_NAMED = {"success": False, "error": "no permission for read on database 'metrics'"} +# The handler gates' own wording, reachable when they check a different +# database/measurement pair than the middleware did. +GATE_SCOPED_ALL = {"error": "access denied: no read permission for database '*'"} +GATE_SCOPED_NAMED = {"error": "access denied: no read permission for database 'metrics'"} +# Arc could not load the token's grants and refused rather than guess. +PERMISSION_DATA_UNAVAILABLE = {"success": False, "error": "permission data unavailable"} +NOT_FOUND = {"error": "Database 'ghost' not found"} + + +@pytest.fixture +def client() -> ArcClient: + c = ArcClient(host="h", token="t", database="db", max_retries=0) + yield c + c.close() + + +# --- Exception hierarchy ----------------------------------------------------- + + +def test_scoped_access_is_a_permission_error() -> None: + """Callers that only care about "refused" can catch one class.""" + assert issubclass(ArcScopedAccessError, ArcPermissionError) + assert issubclass(ArcPermissionError, ArcAuthenticationError) + assert issubclass(ArcPermissionError, ArcError) + + +def test_scoped_access_carries_the_database() -> None: + error = ArcScopedAccessError("refused", database="metrics") + assert error.database == "metrics" + assert ArcScopedAccessError("refused").database is None + + +# --- list() ------------------------------------------------------------------ + + +def test_list_raises_scoped_for_a_tenant_scoped_token( + httpx_mock: HTTPXMock, client: ArcClient +) -> None: + """Arc refuses the list-everything route rather than filtering it.""" + httpx_mock.add_response(status_code=403, json=SCOPED_ALL) + + with pytest.raises(ArcScopedAccessError) as excinfo: + client.databases.list() + + # Arc sends no database name on this route, so the advice must be + # "name one", not "you have no grant for ''". + assert excinfo.value.database == "" + message = str(excinfo.value) + assert "scoped to specific databases" in message + assert "will not list them all" in message + assert "''" not in message.replace("database ''", "") + + +def test_list_raises_permission_error_without_read( + httpx_mock: HTTPXMock, client: ArcClient +) -> None: + """A token with no read permission is a different problem.""" + httpx_mock.add_response(status_code=403, json=NO_READ_PERMISSION) + + with pytest.raises(ArcPermissionError) as excinfo: + client.databases.list() + + assert not isinstance(excinfo.value, ArcScopedAccessError) + assert "Permission denied" in str(excinfo.value) + + +@pytest.mark.parametrize("body", [NO_TOKEN, BAD_TOKEN]) +def test_list_raises_authentication_error_on_401( + httpx_mock: HTTPXMock, client: ArcClient, body: dict +) -> None: + httpx_mock.add_response(status_code=401, json=body) + + with pytest.raises(ArcAuthenticationError) as excinfo: + client.databases.list() + + # 401 must not surface as a permission problem: re-authenticating fixes it, + # a different token's permissions do not. + assert not isinstance(excinfo.value, ArcPermissionError) + assert "Authentication failed" in str(excinfo.value) + + +def test_list_success_unchanged(httpx_mock: HTTPXMock, client: ArcClient) -> None: + httpx_mock.add_response( + json={"databases": [{"name": "metrics", "measurement_count": 2}], "count": 1} + ) + + result = client.databases.list() + + assert result.count == 1 + assert result.databases[0].name == "metrics" + + +# --- get() and list_measurements() ------------------------------------------ + + +@pytest.mark.parametrize("method", ["get", "list_measurements"]) +def test_per_database_routes_raise_scoped_and_name_the_database( + httpx_mock: HTTPXMock, client: ArcClient, method: str +) -> None: + httpx_mock.add_response(status_code=403, json=SCOPED_NAMED) + + with pytest.raises(ArcScopedAccessError) as excinfo: + getattr(client.databases, method)("metrics") + + assert excinfo.value.database == "metrics" + assert "metrics" in str(excinfo.value) + + +@pytest.mark.parametrize("method", ["get", "list_measurements"]) +def test_per_database_routes_keep_404_separate( + httpx_mock: HTTPXMock, client: ArcClient, method: str +) -> None: + """ "You may not read it" and "it is not there" are different answers.""" + httpx_mock.add_response(status_code=404, json=NOT_FOUND) + + with pytest.raises(ArcNotFoundError) as excinfo: + getattr(client.databases, method)("ghost") + + assert not isinstance(excinfo.value, ArcPermissionError) + assert "not found" in str(excinfo.value) + + +@pytest.mark.parametrize("method", ["get", "list_measurements"]) +def test_per_database_routes_raise_permission_error_without_read( + httpx_mock: HTTPXMock, client: ArcClient, method: str +) -> None: + httpx_mock.add_response(status_code=403, json=NO_READ_PERMISSION) + + with pytest.raises(ArcPermissionError) as excinfo: + getattr(client.databases, method)("metrics") + + assert not isinstance(excinfo.value, ArcScopedAccessError) + + +# --- 403s that are not the per-database denial ------------------------------- + + +def test_unknown_403_body_is_not_reported_as_scoped( + httpx_mock: HTTPXMock, client: ArcClient +) -> None: + """Telling a caller to name a database it already named is worse than + naming the permission, so only the recognised shape classifies as scoped.""" + httpx_mock.add_response(status_code=403, json={"error": "Delete operations are disabled."}) + + with pytest.raises(ArcPermissionError) as excinfo: + client.databases.get("metrics") + + assert not isinstance(excinfo.value, ArcScopedAccessError) + + +def test_license_403_still_maps_to_license_error(httpx_mock: HTTPXMock, client: ArcClient) -> None: + """The pre-existing enterprise-license 403 keeps its own exception.""" + from arc_client.exceptions import ArcLicenseError + + httpx_mock.add_response( + status_code=403, + json={"error": "This feature requires an enterprise license"}, + ) + + with pytest.raises(ArcLicenseError): + client.databases.list() + + +# --- async twin -------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_async_list_raises_scoped(httpx_mock: HTTPXMock) -> None: + httpx_mock.add_response(status_code=403, json=SCOPED_ALL) + + c = AsyncArcClient(host="h", token="t", database="db", max_retries=0) + try: + with pytest.raises(ArcScopedAccessError) as excinfo: + await c.databases.list() + assert excinfo.value.database == "" + finally: + await c.close() + + +@pytest.mark.asyncio +async def test_async_list_measurements_raises_scoped_and_names_database( + httpx_mock: HTTPXMock, +) -> None: + httpx_mock.add_response(status_code=403, json=SCOPED_NAMED) + + c = AsyncArcClient(host="h", token="t", database="db", max_retries=0) + try: + with pytest.raises(ArcScopedAccessError) as excinfo: + await c.databases.list_measurements("metrics") + assert excinfo.value.database == "metrics" + finally: + await c.close() + + +# --- the full 403 message matrix --------------------------------------------- + + +@pytest.mark.parametrize( + ("body", "expected", "expected_database"), + [ + # Arc's read middleware, which is what these routes use. + (SCOPED_NAMED, ArcScopedAccessError, "metrics"), + (SCOPED_ALL, ArcScopedAccessError, ""), + # The handler gates' own wording. + (GATE_SCOPED_NAMED, ArcScopedAccessError, "metrics"), + (GATE_SCOPED_ALL, ArcScopedAccessError, "*"), + # The token carries no read permission at all — a different problem, + # and naming a database will not help. + (NO_READ_PERMISSION, ArcPermissionError, None), + (OSS_NO_READ, ArcPermissionError, None), + # Arc could not read its own grants: a server fault, not the token's. + (PERMISSION_DATA_UNAVAILABLE, ArcPermissionDataUnavailableError, None), + # An unrecognised 403 falls back to the permission message, which is + # safe to say about any refusal. + ({"error": "Delete operations are disabled."}, ArcPermissionError, None), + ], +) +def test_every_forbidden_body_maps_to_its_own_exception( + httpx_mock: HTTPXMock, + client: ArcClient, + body: dict, + expected: type, + expected_database: object, +) -> None: + httpx_mock.add_response(status_code=403, json=body) + + with pytest.raises(expected) as excinfo: + client.databases.get("metrics") + + # The mapping must be exact, not just "a subclass of something". + assert type(excinfo.value) is expected + if expected_database is not None: + assert excinfo.value.database == expected_database + + +def test_unavailable_is_not_reported_as_scoped(httpx_mock: HTTPXMock, client: ArcClient) -> None: + """A broken permission store must not send the caller after a database.""" + httpx_mock.add_response(status_code=403, json=PERMISSION_DATA_UNAVAILABLE) + + with pytest.raises(ArcPermissionDataUnavailableError) as excinfo: + client.databases.list() + + assert not isinstance(excinfo.value, ArcScopedAccessError) + message = str(excinfo.value) + assert "server-side fault" in message + assert "scoped" not in message From 1de57d3d174a23ef75f529243ceaa944acece35a Mon Sep 17 00:00:00 2001 From: jallegri <18682329+jallegri@users.noreply.github.com> Date: Tue, 27 Oct 2026 21:56:43 -0300 Subject: [PATCH 2/3] fix(management): align retention and continuous-query contracts --- CHANGELOG.md | 3 + src/arc_client/__init__.py | 2 + src/arc_client/_sentinels.py | 12 + .../management/async_continuous_query.py | 128 ++++++++-- src/arc_client/management/async_retention.py | 85 +++++-- src/arc_client/management/continuous_query.py | 128 ++++++++-- src/arc_client/management/retention.py | 85 +++++-- src/arc_client/models/continuous_query.py | 11 +- src/arc_client/models/retention.py | 16 +- tests/unit/http/test_cq_contracts.py | 145 +++++++++++ tests/unit/http/test_retention_contracts.py | 229 ++++++++++++++++++ 11 files changed, 761 insertions(+), 83 deletions(-) create mode 100644 src/arc_client/_sentinels.py create mode 100644 tests/unit/http/test_cq_contracts.py create mode 100644 tests/unit/http/test_retention_contracts.py diff --git a/CHANGELOG.md b/CHANGELOG.md index ab5948b..4baaefc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -66,6 +66,9 @@ plus bug fixes and the first real HTTP-layer test coverage. ### Fixed +- Retention and continuous-query clients accept Arc's list/null/flat response + shapes. Sparse updates remain sparse; explicit full replacements support + released servers, and CLEAR requests explicit null values on newer servers. - Buffered close waits for active writes, retains unacknowledged batches and surfaces failures without replay. Typed errors retain structured diagnostics. diff --git a/src/arc_client/__init__.py b/src/arc_client/__init__.py index fcbf45b..0118e42 100644 --- a/src/arc_client/__init__.py +++ b/src/arc_client/__init__.py @@ -5,6 +5,7 @@ # failure during package initialization. __version__ = "0.2.0" +from arc_client._sentinels import CLEAR from arc_client.async_client import AsyncArcClient from arc_client.client import ArcClient from arc_client.compression import Compression @@ -27,6 +28,7 @@ ) __all__ = [ + "CLEAR", # Clients "ArcClient", "AsyncArcClient", diff --git a/src/arc_client/_sentinels.py b/src/arc_client/_sentinels.py new file mode 100644 index 0000000..4cf0dce --- /dev/null +++ b/src/arc_client/_sentinels.py @@ -0,0 +1,12 @@ +"""Explicit clearing for nullable fields in partial updates.""" + +from enum import Enum + + +class ClearType(Enum): + """Marker for an explicit JSON null, distinct from an omitted argument.""" + + CLEAR = "CLEAR" + + +CLEAR = ClearType.CLEAR diff --git a/src/arc_client/management/async_continuous_query.py b/src/arc_client/management/async_continuous_query.py index fc3e1fb..e07e5fd 100644 --- a/src/arc_client/management/async_continuous_query.py +++ b/src/arc_client/management/async_continuous_query.py @@ -5,8 +5,9 @@ from datetime import datetime from typing import Any, List, Optional, Union +from arc_client._sentinels import CLEAR, ClearType from arc_client.config import ClientConfig -from arc_client.exceptions import ArcError, ArcNotFoundError +from arc_client.exceptions import ArcError, ArcNotFoundError, ArcValidationError from arc_client.http.async_http import AsyncHTTPClient from arc_client.models.continuous_query import ( ContinuousQuery, @@ -34,6 +35,8 @@ async def create( retention_days: Optional[int] = None, delete_source_after_days: Optional[int] = None, is_active: bool = True, + *, + tag_columns: Optional[List[str]] = None, ) -> ContinuousQuery: """Create a new continuous query.""" payload = { @@ -47,13 +50,14 @@ async def create( "retention_days": retention_days, "delete_source_after_days": delete_source_after_days, "is_active": is_active, + "tag_columns": tag_columns, } try: response = await self._http.post("/api/v1/continuous_queries", json=payload) data = response.json() if not data.get("success", True): raise ArcError(data.get("error", "Failed to create CQ")) - return ContinuousQuery(**data.get("query", data)) + return ContinuousQuery.from_response(data) except ArcError: raise except Exception as e: @@ -72,8 +76,21 @@ async def list( response = await self._http.get("/api/v1/continuous_queries", params=params) data = response.json() - queries = data.get("queries", []) - return [ContinuousQuery(**q) for q in queries] + if data is None: + queries = [] + elif isinstance(data, list): + queries = data + elif isinstance(data, dict) and "queries" in data: + queries = data["queries"] + if queries is None: + queries = [] + else: + raise ArcError("Malformed continuous query list: expected an array") + if not isinstance(queries, list): + raise ArcError("Malformed continuous query list: expected an array") + return [ContinuousQuery.from_response(q) for q in queries] + except ArcError: + raise except Exception as e: raise ArcError(f"Failed to list continuous queries: {e}") from e @@ -87,7 +104,7 @@ async def get(self, query_id: int) -> ContinuousQuery: if "not found" in error.lower(): raise ArcNotFoundError(f"CQ {query_id} not found") raise ArcError(error) - return ContinuousQuery(**data.get("query", data)) + return ContinuousQuery.from_response(data) except (ArcNotFoundError, ArcError): raise except Exception as e: @@ -97,30 +114,85 @@ async def update( self, query_id: int, name: Optional[str] = None, - description: Optional[str] = None, + description: Union[str, ClearType, None] = None, query: Optional[str] = None, interval: Optional[str] = None, - retention_days: Optional[int] = None, - delete_source_after_days: Optional[int] = None, + retention_days: Union[int, ClearType, None] = None, + delete_source_after_days: Union[int, ClearType, None] = None, is_active: Optional[bool] = None, + *, + database: Optional[str] = None, + source_measurement: Optional[str] = None, + destination_measurement: Optional[str] = None, + tag_columns: Union[List[str], ClearType, None] = None, ) -> ContinuousQuery: - """Update a continuous query.""" + """Send a partial update without reading or replacing unrelated fields. + + None omits an argument, preserving existing caller behavior. CLEAR + sends JSON null for nullable fields. Servers with full-replacement + PUT semantics (including Arc 26.09.3) require replace() instead. + """ + values = { + "name": name, + "description": description, + "query": query, + "interval": interval, + "retention_days": retention_days, + "delete_source_after_days": delete_source_after_days, + "is_active": is_active, + "database": database, + "source_measurement": source_measurement, + "destination_measurement": destination_measurement, + "tag_columns": tag_columns, + } + nullable = {"description", "tag_columns", "retention_days", "delete_source_after_days"} payload: dict[str, Any] = {} - if name is not None: - payload["name"] = name - if description is not None: - payload["description"] = description - if query is not None: - payload["query"] = query - if interval is not None: - payload["interval"] = interval - if retention_days is not None: - payload["retention_days"] = retention_days - if delete_source_after_days is not None: - payload["delete_source_after_days"] = delete_source_after_days - if is_active is not None: - payload["is_active"] = is_active + for key, value in values.items(): + if value is CLEAR: + if key not in nullable: + raise ArcValidationError(f"{key} may not be cleared") + payload[key] = None + elif value is not None: + payload[key] = value + return await self._put_definition(query_id, payload) + + async def replace( + self, + query_id: int, + *, + name: str, + database: str, + source_measurement: str, + destination_measurement: str, + query: str, + interval: str, + description: Optional[str], + tag_columns: Optional[List[str]], + retention_days: Optional[int], + delete_source_after_days: Optional[int], + is_active: bool, + ) -> ContinuousQuery: + """Replace all mutable fields with one PUT on old or new servers. + + Every field must be explicit. This operation intentionally overwrites + the definition; it does not fetch or merge a potentially stale copy. + """ + payload = { + "name": name, + "database": database, + "source_measurement": source_measurement, + "destination_measurement": destination_measurement, + "query": query, + "interval": interval, + "description": description, + "tag_columns": tag_columns, + "retention_days": retention_days, + "delete_source_after_days": delete_source_after_days, + "is_active": is_active, + } + return await self._put_definition(query_id, payload) + async def _put_definition(self, query_id: int, payload: dict[str, Any]) -> ContinuousQuery: try: response = await self._http.put(f"/api/v1/continuous_queries/{query_id}", json=payload) data = response.json() @@ -129,8 +201,8 @@ async def update( if "not found" in error.lower(): raise ArcNotFoundError(f"CQ {query_id} not found") raise ArcError(error) - return ContinuousQuery(**data.get("query", data)) - except (ArcNotFoundError, ArcError): + return ContinuousQuery.from_response(data) + except ArcError: raise except Exception as e: raise ArcError(f"Failed to update continuous query: {e}") from e @@ -186,6 +258,12 @@ async def get_executions(self, query_id: int, limit: int = 50) -> List[CQExecuti ) data = response.json() executions = data.get("executions", []) + if executions is None: + executions = [] + if not isinstance(executions, list): + raise ArcError("Malformed CQ history: expected an array") return [CQExecution(**e) for e in executions] + except ArcError: + raise except Exception as e: raise ArcError(f"Failed to get CQ executions: {e}") from e diff --git a/src/arc_client/management/async_retention.py b/src/arc_client/management/async_retention.py index 225a8d2..4229b08 100644 --- a/src/arc_client/management/async_retention.py +++ b/src/arc_client/management/async_retention.py @@ -2,10 +2,11 @@ from __future__ import annotations -from typing import Any, List, Optional +from typing import Any, List, Optional, Union +from arc_client._sentinels import CLEAR, ClearType from arc_client.config import ClientConfig -from arc_client.exceptions import ArcError, ArcNotFoundError +from arc_client.exceptions import ArcError, ArcNotFoundError, ArcValidationError from arc_client.http.async_http import AsyncHTTPClient from arc_client.models.retention import ( ExecuteRetentionResponse, @@ -55,8 +56,14 @@ async def list(self) -> List[RetentionPolicy]: try: response = await self._http.get("/api/v1/retention") data = response.json() - policies = data.get("policies", []) + policies = data.get("policies", data) if isinstance(data, dict) else data + if policies is None: + policies = [] + if not isinstance(policies, list): + raise ArcError("Malformed retention policy list: expected an array") return [RetentionPolicy(**p) for p in policies] + except ArcError: + raise except Exception as e: raise ArcError(f"Failed to list retention policies: {e}") from e @@ -81,23 +88,63 @@ async def update( policy_id: int, name: Optional[str] = None, retention_days: Optional[int] = None, - measurement: Optional[str] = None, + measurement: Union[str, ClearType, None] = None, buffer_days: Optional[int] = None, is_active: Optional[bool] = None, + *, + database: Optional[str] = None, ) -> RetentionPolicy: - """Update a retention policy.""" + """Send a partial update without reading unrelated fields. + + None omits an argument, preserving existing caller behavior. CLEAR + sends JSON null for measurement. Servers with full-replacement PUT + semantics (including Arc 26.09.3) require replace() instead. + """ + values = { + "name": name, + "retention_days": retention_days, + "measurement": measurement, + "buffer_days": buffer_days, + "is_active": is_active, + "database": database, + } payload: dict[str, Any] = {} - if name is not None: - payload["name"] = name - if retention_days is not None: - payload["retention_days"] = retention_days - if measurement is not None: - payload["measurement"] = measurement - if buffer_days is not None: - payload["buffer_days"] = buffer_days - if is_active is not None: - payload["is_active"] = is_active + for key, value in values.items(): + if value is CLEAR: + if key != "measurement": + raise ArcValidationError(f"{key} may not be cleared") + payload[key] = None + elif value is not None: + payload[key] = value + return await self._put_definition(policy_id, payload) + async def replace( + self, + policy_id: int, + *, + name: str, + database: str, + retention_days: int, + measurement: Optional[str], + buffer_days: int, + is_active: bool, + ) -> RetentionPolicy: + """Replace all mutable fields with one PUT on old or new servers. + + Every field must be explicit. This operation intentionally overwrites + the definition; it does not fetch or merge a potentially stale copy. + """ + payload = { + "name": name, + "database": database, + "retention_days": retention_days, + "measurement": measurement, + "buffer_days": buffer_days, + "is_active": is_active, + } + return await self._put_definition(policy_id, payload) + + async def _put_definition(self, policy_id: int, payload: dict[str, Any]) -> RetentionPolicy: try: response = await self._http.put(f"/api/v1/retention/{policy_id}", json=payload) data = response.json() @@ -136,6 +183,8 @@ async def execute( response = await self._http.post(f"/api/v1/retention/{policy_id}/execute", json=payload) data = response.json() return ExecuteRetentionResponse(**data) + except ArcError: + raise except Exception as e: raise ArcError(f"Failed to execute retention policy: {e}") from e @@ -148,6 +197,12 @@ async def get_executions(self, policy_id: int, limit: int = 50) -> List[Retentio ) data = response.json() executions = data.get("executions", []) + if executions is None: + executions = [] + if not isinstance(executions, list): + raise ArcError("Malformed retention execution history: expected an array") return [RetentionExecution(**e) for e in executions] + except ArcError: + raise except Exception as e: raise ArcError(f"Failed to get retention executions: {e}") from e diff --git a/src/arc_client/management/continuous_query.py b/src/arc_client/management/continuous_query.py index 6f0196a..601ae6e 100644 --- a/src/arc_client/management/continuous_query.py +++ b/src/arc_client/management/continuous_query.py @@ -5,8 +5,9 @@ from datetime import datetime from typing import Any, List, Optional, Union +from arc_client._sentinels import CLEAR, ClearType from arc_client.config import ClientConfig -from arc_client.exceptions import ArcError, ArcNotFoundError +from arc_client.exceptions import ArcError, ArcNotFoundError, ArcValidationError from arc_client.http.sync_http import SyncHTTPClient from arc_client.models.continuous_query import ( ContinuousQuery, @@ -34,6 +35,8 @@ def create( retention_days: Optional[int] = None, delete_source_after_days: Optional[int] = None, is_active: bool = True, + *, + tag_columns: Optional[List[str]] = None, ) -> ContinuousQuery: """Create a new continuous query.""" payload = { @@ -47,13 +50,14 @@ def create( "retention_days": retention_days, "delete_source_after_days": delete_source_after_days, "is_active": is_active, + "tag_columns": tag_columns, } try: response = self._http.post("/api/v1/continuous_queries", json=payload) data = response.json() if not data.get("success", True): raise ArcError(data.get("error", "Failed to create CQ")) - return ContinuousQuery(**data.get("query", data)) + return ContinuousQuery.from_response(data) except ArcError: raise except Exception as e: @@ -72,8 +76,21 @@ def list( response = self._http.get("/api/v1/continuous_queries", params=params) data = response.json() - queries = data.get("queries", []) - return [ContinuousQuery(**q) for q in queries] + if data is None: + queries = [] + elif isinstance(data, list): + queries = data + elif isinstance(data, dict) and "queries" in data: + queries = data["queries"] + if queries is None: + queries = [] + else: + raise ArcError("Malformed continuous query list: expected an array") + if not isinstance(queries, list): + raise ArcError("Malformed continuous query list: expected an array") + return [ContinuousQuery.from_response(q) for q in queries] + except ArcError: + raise except Exception as e: raise ArcError(f"Failed to list continuous queries: {e}") from e @@ -87,7 +104,7 @@ def get(self, query_id: int) -> ContinuousQuery: if "not found" in error.lower(): raise ArcNotFoundError(f"CQ {query_id} not found") raise ArcError(error) - return ContinuousQuery(**data.get("query", data)) + return ContinuousQuery.from_response(data) except (ArcNotFoundError, ArcError): raise except Exception as e: @@ -97,30 +114,85 @@ def update( self, query_id: int, name: Optional[str] = None, - description: Optional[str] = None, + description: Union[str, ClearType, None] = None, query: Optional[str] = None, interval: Optional[str] = None, - retention_days: Optional[int] = None, - delete_source_after_days: Optional[int] = None, + retention_days: Union[int, ClearType, None] = None, + delete_source_after_days: Union[int, ClearType, None] = None, is_active: Optional[bool] = None, + *, + database: Optional[str] = None, + source_measurement: Optional[str] = None, + destination_measurement: Optional[str] = None, + tag_columns: Union[List[str], ClearType, None] = None, ) -> ContinuousQuery: - """Update a continuous query.""" + """Send a partial update without reading or replacing unrelated fields. + + None omits an argument, preserving existing caller behavior. CLEAR + sends JSON null for nullable fields. Servers with full-replacement + PUT semantics (including Arc 26.09.3) require replace() instead. + """ + values = { + "name": name, + "description": description, + "query": query, + "interval": interval, + "retention_days": retention_days, + "delete_source_after_days": delete_source_after_days, + "is_active": is_active, + "database": database, + "source_measurement": source_measurement, + "destination_measurement": destination_measurement, + "tag_columns": tag_columns, + } + nullable = {"description", "tag_columns", "retention_days", "delete_source_after_days"} payload: dict[str, Any] = {} - if name is not None: - payload["name"] = name - if description is not None: - payload["description"] = description - if query is not None: - payload["query"] = query - if interval is not None: - payload["interval"] = interval - if retention_days is not None: - payload["retention_days"] = retention_days - if delete_source_after_days is not None: - payload["delete_source_after_days"] = delete_source_after_days - if is_active is not None: - payload["is_active"] = is_active + for key, value in values.items(): + if value is CLEAR: + if key not in nullable: + raise ArcValidationError(f"{key} may not be cleared") + payload[key] = None + elif value is not None: + payload[key] = value + return self._put_definition(query_id, payload) + + def replace( + self, + query_id: int, + *, + name: str, + database: str, + source_measurement: str, + destination_measurement: str, + query: str, + interval: str, + description: Optional[str], + tag_columns: Optional[List[str]], + retention_days: Optional[int], + delete_source_after_days: Optional[int], + is_active: bool, + ) -> ContinuousQuery: + """Replace all mutable fields with one PUT on old or new servers. + + Every field must be explicit. This operation intentionally overwrites + the definition; it does not fetch or merge a potentially stale copy. + """ + payload = { + "name": name, + "database": database, + "source_measurement": source_measurement, + "destination_measurement": destination_measurement, + "query": query, + "interval": interval, + "description": description, + "tag_columns": tag_columns, + "retention_days": retention_days, + "delete_source_after_days": delete_source_after_days, + "is_active": is_active, + } + return self._put_definition(query_id, payload) + def _put_definition(self, query_id: int, payload: dict[str, Any]) -> ContinuousQuery: try: response = self._http.put(f"/api/v1/continuous_queries/{query_id}", json=payload) data = response.json() @@ -129,8 +201,8 @@ def update( if "not found" in error.lower(): raise ArcNotFoundError(f"CQ {query_id} not found") raise ArcError(error) - return ContinuousQuery(**data.get("query", data)) - except (ArcNotFoundError, ArcError): + return ContinuousQuery.from_response(data) + except ArcError: raise except Exception as e: raise ArcError(f"Failed to update continuous query: {e}") from e @@ -186,6 +258,12 @@ def get_executions(self, query_id: int, limit: int = 50) -> List[CQExecution]: ) data = response.json() executions = data.get("executions", []) + if executions is None: + executions = [] + if not isinstance(executions, list): + raise ArcError("Malformed CQ history: expected an array") return [CQExecution(**e) for e in executions] + except ArcError: + raise except Exception as e: raise ArcError(f"Failed to get CQ executions: {e}") from e diff --git a/src/arc_client/management/retention.py b/src/arc_client/management/retention.py index f7cf361..fb0128a 100644 --- a/src/arc_client/management/retention.py +++ b/src/arc_client/management/retention.py @@ -2,10 +2,11 @@ from __future__ import annotations -from typing import Any, List, Optional +from typing import Any, List, Optional, Union +from arc_client._sentinels import CLEAR, ClearType from arc_client.config import ClientConfig -from arc_client.exceptions import ArcError, ArcNotFoundError +from arc_client.exceptions import ArcError, ArcNotFoundError, ArcValidationError from arc_client.http.sync_http import SyncHTTPClient from arc_client.models.retention import ( ExecuteRetentionResponse, @@ -55,8 +56,14 @@ def list(self) -> List[RetentionPolicy]: try: response = self._http.get("/api/v1/retention") data = response.json() - policies = data.get("policies", []) + policies = data.get("policies", data) if isinstance(data, dict) else data + if policies is None: + policies = [] + if not isinstance(policies, list): + raise ArcError("Malformed retention policy list: expected an array") return [RetentionPolicy(**p) for p in policies] + except ArcError: + raise except Exception as e: raise ArcError(f"Failed to list retention policies: {e}") from e @@ -81,23 +88,63 @@ def update( policy_id: int, name: Optional[str] = None, retention_days: Optional[int] = None, - measurement: Optional[str] = None, + measurement: Union[str, ClearType, None] = None, buffer_days: Optional[int] = None, is_active: Optional[bool] = None, + *, + database: Optional[str] = None, ) -> RetentionPolicy: - """Update a retention policy.""" + """Send a partial update without reading unrelated fields. + + None omits an argument, preserving existing caller behavior. CLEAR + sends JSON null for measurement. Servers with full-replacement PUT + semantics (including Arc 26.09.3) require replace() instead. + """ + values = { + "name": name, + "retention_days": retention_days, + "measurement": measurement, + "buffer_days": buffer_days, + "is_active": is_active, + "database": database, + } payload: dict[str, Any] = {} - if name is not None: - payload["name"] = name - if retention_days is not None: - payload["retention_days"] = retention_days - if measurement is not None: - payload["measurement"] = measurement - if buffer_days is not None: - payload["buffer_days"] = buffer_days - if is_active is not None: - payload["is_active"] = is_active + for key, value in values.items(): + if value is CLEAR: + if key != "measurement": + raise ArcValidationError(f"{key} may not be cleared") + payload[key] = None + elif value is not None: + payload[key] = value + return self._put_definition(policy_id, payload) + def replace( + self, + policy_id: int, + *, + name: str, + database: str, + retention_days: int, + measurement: Optional[str], + buffer_days: int, + is_active: bool, + ) -> RetentionPolicy: + """Replace all mutable fields with one PUT on old or new servers. + + Every field must be explicit. This operation intentionally overwrites + the definition; it does not fetch or merge a potentially stale copy. + """ + payload = { + "name": name, + "database": database, + "retention_days": retention_days, + "measurement": measurement, + "buffer_days": buffer_days, + "is_active": is_active, + } + return self._put_definition(policy_id, payload) + + def _put_definition(self, policy_id: int, payload: dict[str, Any]) -> RetentionPolicy: try: response = self._http.put(f"/api/v1/retention/{policy_id}", json=payload) data = response.json() @@ -136,6 +183,8 @@ def execute( response = self._http.post(f"/api/v1/retention/{policy_id}/execute", json=payload) data = response.json() return ExecuteRetentionResponse(**data) + except ArcError: + raise except Exception as e: raise ArcError(f"Failed to execute retention policy: {e}") from e @@ -148,6 +197,12 @@ def get_executions(self, policy_id: int, limit: int = 50) -> List[RetentionExecu ) data = response.json() executions = data.get("executions", []) + if executions is None: + executions = [] + if not isinstance(executions, list): + raise ArcError("Malformed retention execution history: expected an array") return [RetentionExecution(**e) for e in executions] + except ArcError: + raise except Exception as e: raise ArcError(f"Failed to get retention executions: {e}") from e diff --git a/src/arc_client/models/continuous_query.py b/src/arc_client/models/continuous_query.py index 04909d2..de435ff 100644 --- a/src/arc_client/models/continuous_query.py +++ b/src/arc_client/models/continuous_query.py @@ -2,7 +2,7 @@ from __future__ import annotations -from typing import Optional +from typing import Any, List, Optional from pydantic import BaseModel @@ -18,6 +18,7 @@ class ContinuousQuery(BaseModel): destination_measurement: str query: str interval: str + tag_columns: Optional[List[str]] = None retention_days: Optional[int] = None delete_source_after_days: Optional[int] = None is_active: bool = True @@ -28,6 +29,14 @@ class ContinuousQuery(BaseModel): created_at: Optional[str] = None updated_at: Optional[str] = None + @classmethod + def from_response(cls, data: dict[str, Any]) -> ContinuousQuery: + """Accept flat definitions and legacy mapping envelopes, never SQL text.""" + if not isinstance(data, dict): + raise ValueError("Expected a continuous query object") + nested = data.get("query") + return cls(**nested) if isinstance(nested, dict) else cls(**data) + class ExecuteCQResponse(BaseModel): """Response from executing a continuous query.""" diff --git a/src/arc_client/models/retention.py b/src/arc_client/models/retention.py index 4fbfc45..ecfeb0f 100644 --- a/src/arc_client/models/retention.py +++ b/src/arc_client/models/retention.py @@ -2,9 +2,9 @@ from __future__ import annotations -from typing import Optional +from typing import Any, Optional -from pydantic import BaseModel, Field +from pydantic import BaseModel, Field, field_validator class RetentionPolicy(BaseModel): @@ -31,12 +31,24 @@ class ExecuteRetentionResponse(BaseModel): policy_name: str deleted_count: int = 0 files_deleted: int = 0 + skipped_files: int = 0 + skipped_reason: str = "" + cold_files_deleted: int = 0 + cold_rows_deleted: int = 0 + cold_files_skipped: int = 0 + cold_skipped_reason: str = "" execution_time_ms: float = 0.0 dry_run: bool = False cutoff_date: Optional[str] = None affected_measurements: list[str] = Field(default_factory=list) error: Optional[str] = None + @field_validator("affected_measurements", mode="before") + @classmethod + def normalize_measurements(cls, value: Any) -> Any: + """Treat a nil Go slice as empty, preserving validation for other values.""" + return [] if value is None else value + class RetentionExecution(BaseModel): """A retention execution history record.""" diff --git a/tests/unit/http/test_cq_contracts.py b/tests/unit/http/test_cq_contracts.py new file mode 100644 index 0000000..5fe63e5 --- /dev/null +++ b/tests/unit/http/test_cq_contracts.py @@ -0,0 +1,145 @@ +"""Continuous-query wire contracts shared by sync and async public clients.""" + +import inspect +import json +from contextlib import AsyncExitStack + +import pytest + +from arc_client import CLEAR, ArcClient, AsyncArcClient +from arc_client.exceptions import ArcPermissionError, ArcValidationError + +DEFINITION = { + "id": 7, + "name": "rollup", + "database": "metrics", + "source_measurement": "cpu", + "destination_measurement": "cpu_hour", + "query": "SELECT avg(value) FROM metrics.cpu", + "interval": "1h", + "description": "keep", + "tag_columns": ["host"], + "retention_days": 30, + "delete_source_after_days": 60, + "is_active": True, +} + + +async def call(client, method, *args, **kwargs): + result = getattr(client.continuous_queries, method)(*args, **kwargs) + return await result if inspect.isawaitable(result) else result + + +@pytest.fixture(params=[False, True], ids=["sync", "async"]) +def async_mode(request): + return request.param + + +async def enter_client(stack, async_mode): + if async_mode: + return await stack.enter_async_context(AsyncArcClient(token="test", max_retries=0)) + return stack.enter_context(ArcClient(token="test", max_retries=0)) + + +@pytest.mark.parametrize("body", [None, [], {"queries": None}, {"queries": []}]) +async def test_empty_list_shapes(httpx_mock, async_mode, body): + httpx_mock.add_response( + method="GET", + content=json.dumps(body).encode(), + headers={"Content-Type": "application/json"}, + ) + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + assert await call(client, "list") == [] + + +async def test_empty_execution_history(httpx_mock, async_mode): + httpx_mock.add_response(method="GET", json={"executions": None}) + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + assert await call(client, "get_executions", 7) == [] + + +async def test_create_flat_definition_and_tags(httpx_mock, async_mode): + httpx_mock.add_response(method="POST", json=DEFINITION) + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + args = { + k: DEFINITION[k] + for k in ( + "name", + "database", + "source_measurement", + "destination_measurement", + "query", + "interval", + "tag_columns", + ) + } + result = await call(client, "create", **args) + assert result.query == DEFINITION["query"] + assert result.tag_columns == ["host"] + assert json.loads(httpx_mock.get_request().content)["tag_columns"] == ["host"] + + +async def test_sparse_update_no_get_preserves_false_zero_and_clear(httpx_mock, async_mode): + httpx_mock.add_response( + method="PUT", json={**DEFINITION, "is_active": False, "description": None} + ) + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + result = await call( + client, + "update", + 7, + is_active=False, + description=CLEAR, + retention_days=0, + tag_columns=[], + database=None, + ) + assert not result.is_active + requests = httpx_mock.get_requests() + assert len(requests) == 1 + assert requests[0].method == "PUT" + assert json.loads(requests[0].content) == { + "is_active": False, + "description": None, + "retention_days": 0, + "tag_columns": [], + } + + +async def test_replace_sends_complete_definition_once(httpx_mock, async_mode): + httpx_mock.add_response(method="PUT", json=DEFINITION) + args = {k: v for k, v in DEFINITION.items() if k != "id"} + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + result = await call(client, "replace", 7, **args) + assert result.tag_columns == ["host"] + requests = httpx_mock.get_requests() + assert len(requests) == 1 + assert json.loads(requests[0].content) == args + + +async def test_none_omits_existing_arguments(httpx_mock, async_mode): + httpx_mock.add_response(method="PUT", json=DEFINITION) + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + await call(client, "update", 7, description=None, retention_days=None) + assert json.loads(httpx_mock.get_request().content) == {} + + +async def test_permissions_remain_typed(httpx_mock, async_mode): + httpx_mock.add_response(status_code=403, json={"error": "Permission denied: read required"}) + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + with pytest.raises(ArcPermissionError): + await call(client, "list") + + +async def test_clear_nonnullable_rejected_before_request(async_mode): + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + with pytest.raises(ArcValidationError): + await call(client, "update", 7, name=CLEAR) diff --git a/tests/unit/http/test_retention_contracts.py b/tests/unit/http/test_retention_contracts.py new file mode 100644 index 0000000..ba73606 --- /dev/null +++ b/tests/unit/http/test_retention_contracts.py @@ -0,0 +1,229 @@ +"""Retention wire contracts exercised through both public clients.""" + +import inspect +import json +from contextlib import AsyncExitStack + +import pytest + +from arc_client import CLEAR, ArcClient, AsyncArcClient +from arc_client.exceptions import ArcError, ArcPermissionError, ArcValidationError + +DEFINITION = { + "id": 7, + "name": "cleanup", + "database": "metrics", + "measurement": "cpu", + "retention_days": 30, + "buffer_days": 7, + "is_active": True, +} +REPLACEMENT = {key: value for key, value in DEFINITION.items() if key != "id"} + + +async def call(client, method, *args, **kwargs): + result = getattr(client.retention, method)(*args, **kwargs) + return await result if inspect.isawaitable(result) else result + + +@pytest.fixture(params=[False, True], ids=["sync", "async"]) +def async_mode(request): + return request.param + + +async def enter_client(stack, async_mode): + if async_mode: + return await stack.enter_async_context(AsyncArcClient(token="test", max_retries=0)) + return stack.enter_context(ArcClient(token="test", max_retries=0)) + + +@pytest.mark.parametrize("body", [None, [], {"policies": None}, {"policies": []}]) +async def test_empty_list_shapes(httpx_mock, async_mode, body): + httpx_mock.add_response(content=json.dumps(body).encode()) + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + assert await call(client, "list") == [] + + +@pytest.mark.parametrize("body", [[DEFINITION], {"policies": [DEFINITION]}]) +async def test_populated_list_shapes(httpx_mock, async_mode, body): + httpx_mock.add_response(json=body) + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + assert (await call(client, "list"))[0].measurement == "cpu" + + +@pytest.mark.parametrize("body", [False, 0, "", {}, {"policies": False}, {"policies": {}}]) +async def test_malformed_lists_rejected(httpx_mock, async_mode, body): + httpx_mock.add_response(json=body) + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + with pytest.raises(ArcError): + await call(client, "list") + + +@pytest.mark.parametrize("executions", [None, []]) +async def test_empty_history(httpx_mock, async_mode, executions): + httpx_mock.add_response(json={"executions": executions}) + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + assert await call(client, "get_executions", 7, limit=3) == [] + assert httpx_mock.get_request().url.params["limit"] == "3" + + +@pytest.mark.parametrize("executions", [False, 0, "", {}]) +async def test_malformed_history_rejected(httpx_mock, async_mode, executions): + httpx_mock.add_response(json={"executions": executions}) + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + with pytest.raises(ArcError): + await call(client, "get_executions", 7) + + +@pytest.mark.parametrize("measurement", [CLEAR, ""]) +async def test_sparse_update_single_put(httpx_mock, async_mode, measurement): + httpx_mock.add_response(method="PUT", json={**DEFINITION, "is_active": False}) + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + result = await call( + client, + "update", + 7, + name="", + retention_days=0, + measurement=measurement, + buffer_days=0, + is_active=False, + database="other", + ) + assert result.is_active is False + requests = httpx_mock.get_requests() + assert len(requests) == 1 + assert requests[0].method == "PUT" + assert requests[0].url.path == "/api/v1/retention/7" + assert json.loads(requests[0].content) == { + "name": "", + "retention_days": 0, + "measurement": None if measurement is CLEAR else "", + "buffer_days": 0, + "is_active": False, + "database": "other", + } + + +async def test_none_omits_and_positional_arguments_stay_compatible(httpx_mock, async_mode): + httpx_mock.add_response(method="PUT", json={"policy": DEFINITION}) + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + await call(client, "update", 7, None, None, None, None, None, database=None) + assert json.loads(httpx_mock.get_request().content) == {} + + +@pytest.mark.parametrize( + "field", ["name", "database", "retention_days", "buffer_days", "is_active"] +) +async def test_clear_nonnullable_rejected_before_http(httpx_mock, async_mode, field): + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + with pytest.raises(ArcValidationError): + await call(client, "update", 7, **{field: CLEAR}) + assert httpx_mock.get_requests() == [] + + +@pytest.mark.parametrize("measurement", [None, "cpu"]) +async def test_replace_complete_single_put(httpx_mock, async_mode, measurement): + values = {**REPLACEMENT, "measurement": measurement, "buffer_days": 0, "is_active": False} + httpx_mock.add_response(method="PUT", json={"id": 7, **values}) + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + result = await call(client, "replace", 7, **values) + assert result.measurement == measurement + requests = httpx_mock.get_requests() + assert len(requests) == 1 + assert requests[0].method == "PUT" + assert json.loads(requests[0].content) == values + + +@pytest.mark.parametrize("field", list(REPLACEMENT)) +async def test_replace_requires_every_field(httpx_mock, async_mode, field): + values = {key: value for key, value in REPLACEMENT.items() if key != field} + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + with pytest.raises(TypeError): + await call(client, "replace", 7, **values) + assert httpx_mock.get_requests() == [] + + +@pytest.mark.parametrize( + "method,args,kwargs", + [ + ("create", (), REPLACEMENT), + ("get", (7,), {}), + ("list", (), {}), + ("update", (7,), {"is_active": False}), + ("replace", (7,), REPLACEMENT), + ("delete", (7,), {}), + ("execute", (7,), {}), + ("get_executions", (7,), {}), + ], +) +async def test_permissions_remain_typed_without_fallback( + httpx_mock, async_mode, method, args, kwargs +): + httpx_mock.add_response(status_code=403, json={"error": "Permission denied: read required"}) + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + with pytest.raises(ArcPermissionError): + await call(client, method, *args, **kwargs) + assert len(httpx_mock.get_requests()) == 1 + + +async def test_execution_diagnostics_and_null_measurements(httpx_mock, async_mode): + diagnostics = { + "skipped_files": 2, + "skipped_reason": "unaddressable files", + "cold_files_deleted": 3, + "cold_rows_deleted": 42, + "cold_files_skipped": 4, + "cold_skipped_reason": "cold tier unavailable", + } + httpx_mock.add_response( + json={ + "policy_id": 7, + "policy_name": "cleanup", + "affected_measurements": None, + "dry_run": True, + **diagnostics, + } + ) + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + result = await call(client, "execute", 7) + assert result.affected_measurements == [] + assert {field: getattr(result, field) for field in diagnostics} == diagnostics + assert json.loads(httpx_mock.get_request().content) == {"dry_run": True, "confirm": False} + + +async def test_older_execution_defaults(httpx_mock, async_mode): + httpx_mock.add_response(json={"policy_id": 7, "policy_name": "cleanup"}) + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + result = await call(client, "execute", 7) + assert result.skipped_files == result.cold_files_deleted == result.cold_rows_deleted == 0 + assert result.cold_files_skipped == 0 + assert result.skipped_reason == result.cold_skipped_reason == "" + + +@pytest.mark.parametrize("measurements", [False, 0, "", {}]) +async def test_malformed_measurements_rejected(httpx_mock, async_mode, measurements): + httpx_mock.add_response( + json={ + "policy_id": 7, + "policy_name": "cleanup", + "affected_measurements": measurements, + } + ) + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + with pytest.raises(ArcError): + await call(client, "execute", 7) From f56bb9f1f82ba1369163746b0e9dfd9e2589e4dd Mon Sep 17 00:00:00 2001 From: jallegri <18682329+jallegri@users.noreply.github.com> Date: Tue, 27 Oct 2026 21:56:43 -0300 Subject: [PATCH 3/3] fix(http): preserve mutation safety and buffered batches --- .github/workflows/ci.yml | 24 +- CHANGELOG.md | 12 +- src/arc_client/config.py | 4 +- src/arc_client/exceptions.py | 24 +- src/arc_client/http/async_http.py | 21 +- src/arc_client/http/base.py | 16 +- src/arc_client/http/sync_http.py | 21 +- src/arc_client/ingestion/async_buffered.py | 61 +++-- src/arc_client/ingestion/buffered.py | 62 +++-- tests/unit/http/test_error_diagnostics.py | 38 +++ tests/unit/http/test_retry_safety.py | 67 +++++ tests/unit/test_buffered_lifecycle.py | 286 +++++++++++++++++++++ 12 files changed, 572 insertions(+), 64 deletions(-) create mode 100644 tests/unit/http/test_error_diagnostics.py create mode 100644 tests/unit/http/test_retry_safety.py create mode 100644 tests/unit/test_buffered_lifecycle.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7b0b81b..9353145 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -4,9 +4,11 @@ on: push: branches: [main] pull_request: - branches: [main] workflow_call: # Allow release.yml to call this workflow +env: + UV_PYTHON: "3.11" + jobs: lint: name: Lint @@ -23,7 +25,7 @@ jobs: run: uv python install 3.11 - name: Install dependencies - run: uv sync --extra dev + run: uv sync --extra dev --extra all - name: Run ruff check run: uv run ruff check . @@ -46,13 +48,15 @@ jobs: run: uv python install 3.11 - name: Install dependencies - run: uv sync --extra dev + run: uv sync --extra dev --extra all - name: Run mypy run: uv run mypy src/arc_client --ignore-missing-imports test: name: Test (Python ${{ matrix.python-version }}) + env: + UV_PYTHON: ${{ matrix.python-version }} runs-on: ubuntu-latest strategy: fail-fast: false @@ -71,10 +75,12 @@ jobs: run: uv python install ${{ matrix.python-version }} - name: Install dependencies - run: uv sync --extra dev + run: uv sync --extra dev --extra all - name: Run tests - run: uv run pytest tests/unit -v --tb=short + run: | + uv run python --version + uv run pytest tests/unit -v --tb=short test-platforms: name: Test (${{ matrix.os }}) @@ -96,10 +102,12 @@ jobs: run: uv python install 3.11 - name: Install dependencies - run: uv sync --extra dev + run: uv sync --extra dev --extra all - name: Run tests - run: uv run pytest tests/unit -v --tb=short + run: | + uv run python --version + uv run pytest tests/unit -v --tb=short coverage: name: Coverage @@ -116,7 +124,7 @@ jobs: run: uv python install 3.11 - name: Install dependencies - run: uv sync --extra dev + run: uv sync --extra dev --extra all - name: Run tests with coverage run: uv run pytest tests/unit --cov=src/arc_client --cov-report=xml diff --git a/CHANGELOG.md b/CHANGELOG.md index 447d024..ab5948b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,7 +5,7 @@ All notable changes to `arc-tsdb-client` are documented here. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). -## [0.2.0] - 2026-07-03 +## [0.2.0] - Unreleased A large feature-parity release bringing the SDK up to date with the Arc server, plus bug fixes and the first real HTTP-layer test coverage. @@ -28,9 +28,10 @@ plus bug fixes and the first real HTTP-layer test coverage. `mqtt`. - **zstd write compression** (now the default; gzip and none still supported). Compression is configurable via a `Compression` enum, a string, or a bool. -- **Automatic retries** with exponential backoff + jitter for connection errors, - 429 (honoring `Retry-After`), and 503. Configurable via `max_retries` / - `retry_backoff`. +- **Automatic retries** with exponential backoff + jitter for read-only + connection/transport failures, 429 (honoring `Retry-After`) and 503. Mutations + retry only connection establishment; data DELETE never retries. Configure + via `max_retries` / `retry_backoff`. - New exceptions: `ArcNotSupportedError` (501) and `ArcLicenseError` (403 for a missing enterprise-license feature; subclass of `ArcAuthenticationError`). - `ArcPermissionError` (403: the token lacks the permission the endpoint @@ -65,6 +66,9 @@ plus bug fixes and the first real HTTP-layer test coverage. ### Fixed +- Buffered close waits for active writes, retains unacknowledged batches and + surfaces failures without replay. Typed errors retain structured diagnostics. + - Polars datetime → columnar conversion used a fragile private attribute that raised at runtime; it now casts via the public `polars` API and normalizes the series' `time_unit` to microseconds. diff --git a/src/arc_client/config.py b/src/arc_client/config.py index c3f7c83..0e4caae 100644 --- a/src/arc_client/config.py +++ b/src/arc_client/config.py @@ -27,8 +27,8 @@ class ClientConfig(BaseModel): max_retries: int = Field( default=3, ge=0, - description="Max automatic retries for transient failures " - "(connection errors, 429, 503). 0 disables retries.", + description="Max retries for read-only transient failures or mutation connection " + "establishment failures. Data DELETE never retries. 0 disables retries.", ) retry_backoff: float = Field( default=0.5, diff --git a/src/arc_client/exceptions.py b/src/arc_client/exceptions.py index a0244ff..fad56fd 100644 --- a/src/arc_client/exceptions.py +++ b/src/arc_client/exceptions.py @@ -2,13 +2,29 @@ from __future__ import annotations -from typing import Optional +from typing import Any, Optional class ArcError(Exception): - """Base exception for arc-client.""" - - pass + """Base exception; HTTP failures retain structured diagnostics when available.""" + + _response_body: Optional[dict[str, Any]] = None + + @property + def response_body(self) -> Optional[dict[str, Any]]: + """Retain diagnostics through existing higher-level exception wrappers.""" + error: Optional[BaseException] = self + seen: set[int] = set() + while isinstance(error, ArcError) and id(error) not in seen: + seen.add(id(error)) + if error._response_body is not None: + return error._response_body + error = error.__cause__ + return None + + @response_body.setter + def response_body(self, value: Optional[dict[str, Any]]) -> None: + self._response_body = value class ArcConnectionError(ArcError): diff --git a/src/arc_client/http/async_http.py b/src/arc_client/http/async_http.py index cb9e32d..bb6598c 100644 --- a/src/arc_client/http/async_http.py +++ b/src/arc_client/http/async_http.py @@ -52,24 +52,37 @@ async def _request( ) -> httpx.Response: """Send a request with retry/backoff for transient failures. - Retries connection/timeout errors and 429/503 responses up to - ``config.max_retries`` times, honoring ``Retry-After`` when present. + Reads retry transient failures up to ``config.max_retries`` times. + Mutations retry only connection establishment failures; data DELETE + never retries. Unknown write outcomes are returned to the caller. """ request_kwargs = self._prepare_request_kwargs(headers=headers, **kwargs) client = self._get_client() last_exc: Optional[Exception] = None + read_only = method.upper() in {"GET", "HEAD", "OPTIONS"} + # DELETE may partially commit. Other mutations may be retried only + # when no application request could have reached the server. + data_delete = method.upper() == "POST" and path.rstrip("/") == "/api/v1/delete" for attempt in range(self.config.max_retries + 1): try: response = await client.request(method, path, **request_kwargs) except RETRYABLE_TRANSPORT_ERRORS as e: last_exc = e - if attempt >= self.config.max_retries: + if ( + data_delete + or not (read_only or isinstance(e, (httpx.ConnectError, httpx.ConnectTimeout))) + or attempt >= self.config.max_retries + ): break await asyncio.sleep(compute_backoff(attempt, self.config.retry_backoff)) continue - if is_retryable_status(response.status_code) and attempt < self.config.max_retries: + if ( + read_only + and is_retryable_status(response.status_code) + and attempt < self.config.max_retries + ): retry_after = parse_retry_after(response) await asyncio.sleep( compute_backoff(attempt, self.config.retry_backoff, retry_after) diff --git a/src/arc_client/http/base.py b/src/arc_client/http/base.py index ffa588b..1cb8a3f 100644 --- a/src/arc_client/http/base.py +++ b/src/arc_client/http/base.py @@ -13,6 +13,7 @@ from arc_client.exceptions import ( ArcAuthenticationError, ArcConnectionError, + ArcError, ArcLicenseError, ArcNotFoundError, ArcNotSupportedError, @@ -100,6 +101,19 @@ def _is_permission_data_unavailable(message: object) -> bool: def handle_response_error(response: httpx.Response) -> None: + """Raise the existing typed error, retaining the server's diagnostic object.""" + try: + _raise_response_error(response) + except ArcError as error: + try: + body = response.json() + except ValueError: + body = None + error.response_body = body if isinstance(body, dict) else None + raise + + +def _raise_response_error(response: httpx.Response) -> None: """Handle HTTP error responses and raise appropriate exceptions. Raises: @@ -185,7 +199,7 @@ def handle_connection_error(error: Exception, url: str) -> None: raise ArcConnectionError(f"Failed to connect to {url}: {error}") from error -# Transport errors that are safe to retry (connection refused, DNS, timeouts). +# Transient transport errors eligible for retry on read-only requests. RETRYABLE_TRANSPORT_ERRORS = ( httpx.ConnectError, httpx.ConnectTimeout, diff --git a/src/arc_client/http/sync_http.py b/src/arc_client/http/sync_http.py index b4beb7c..2e46e52 100644 --- a/src/arc_client/http/sync_http.py +++ b/src/arc_client/http/sync_http.py @@ -52,24 +52,37 @@ def _request( ) -> httpx.Response: """Send a request with retry/backoff for transient failures. - Retries connection/timeout errors and 429/503 responses up to - ``config.max_retries`` times, honoring ``Retry-After`` when present. + Reads retry transient failures up to ``config.max_retries`` times. + Mutations retry only connection establishment failures; data DELETE + never retries. Unknown write outcomes are returned to the caller. """ request_kwargs = self._prepare_request_kwargs(headers=headers, **kwargs) client = self._get_client() last_exc: Optional[Exception] = None + read_only = method.upper() in {"GET", "HEAD", "OPTIONS"} + # DELETE may partially commit. Other mutations may be retried only + # when no application request could have reached the server. + data_delete = method.upper() == "POST" and path.rstrip("/") == "/api/v1/delete" for attempt in range(self.config.max_retries + 1): try: response = client.request(method, path, **request_kwargs) except RETRYABLE_TRANSPORT_ERRORS as e: last_exc = e - if attempt >= self.config.max_retries: + if ( + data_delete + or not (read_only or isinstance(e, (httpx.ConnectError, httpx.ConnectTimeout))) + or attempt >= self.config.max_retries + ): break time.sleep(compute_backoff(attempt, self.config.retry_backoff)) continue - if is_retryable_status(response.status_code) and attempt < self.config.max_retries: + if ( + read_only + and is_retryable_status(response.status_code) + and attempt < self.config.max_retries + ): retry_after = parse_retry_after(response) time.sleep(compute_backoff(attempt, self.config.retry_backoff, retry_after)) continue diff --git a/src/arc_client/ingestion/async_buffered.py b/src/arc_client/ingestion/async_buffered.py index c984115..6f5c582 100644 --- a/src/arc_client/ingestion/async_buffered.py +++ b/src/arc_client/ingestion/async_buffered.py @@ -3,7 +3,6 @@ from __future__ import annotations import asyncio -import contextlib import time from collections import defaultdict from typing import TYPE_CHECKING, Any @@ -38,9 +37,12 @@ def __init__( self._last_flush_time = time.monotonic() self._lock = asyncio.Lock() self._closed = False + self._closing = False + self._flush_error: BaseException | None = None # Background task that flushes on flush_interval even when idle. self._flush_task: asyncio.Task[None] | None = None + self._stop_event = asyncio.Event() async def write(self, record: dict[str, Any]) -> None: """Write a single record to the buffer.""" @@ -65,6 +67,9 @@ async def write(self, record: dict[str, Any]) -> None: columns[key] = [value] async with self._lock: + self._raise_if_failed() + if self._closed or self._closing: + raise RuntimeError("Buffered writer is closed") self._buffers[measurement].append(columns) self._record_counts[measurement] += 1 @@ -86,6 +91,9 @@ async def write_columnar( num_records = len(next(iter(columns.values()))) async with self._lock: + self._raise_if_failed() + if self._closed or self._closing: + raise RuntimeError("Buffered writer is closed") self._buffers[measurement].append(columns) self._record_counts[measurement] += num_records @@ -96,25 +104,40 @@ async def write_columnar( await self._flush_all_unlocked() async def flush(self) -> None: - """Manually flush all buffered data.""" + """Flush all buffered data, waiting for any in-flight write. + + A failed write leaves its data pending and makes the buffer terminal: + subsequent flush/close calls raise that failure without retrying data + whose server outcome may be unknown. Producers wait while I/O runs. + """ async with self._lock: await self._flush_all_unlocked() async def _flush_loop(self) -> None: """Periodically flush measurements whose data has aged past the interval.""" try: - while True: - await asyncio.sleep(self._flush_interval) + while not self._stop_event.is_set(): + try: + await asyncio.wait_for(self._stop_event.wait(), self._flush_interval) + return + except asyncio.TimeoutError: + pass async with self._lock: - if self._closed: + if self._closed or self._stop_event.is_set(): return if ( self._buffers and time.monotonic() - self._last_flush_time >= self._flush_interval ): await self._flush_all_unlocked() - except asyncio.CancelledError: - return + except Exception as error: + # Foreground flush/close will surface this error. Do not retry an + # uncertain outcome, or leave a task exception unobserved. + self._flush_error = error + + def _raise_if_failed(self) -> None: + if self._flush_error is not None: + raise self._flush_error async def _flush_measurement_unlocked(self, measurement: str) -> None: """Flush a single measurement. Must hold lock.""" @@ -122,20 +145,21 @@ async def _flush_measurement_unlocked(self, measurement: str) -> None: return merged = self._merge_columnar(self._buffers[measurement]) - self._buffers[measurement] = [] - self._record_counts[measurement] = 0 - - # Release lock during I/O - self._lock.release() + # Retain data and the lock until acknowledgement. Concurrent timer, + # manual, and close flushes must never send the same batch twice. try: await self._client.write_columnar(measurement, merged) - finally: - await self._lock.acquire() + except BaseException as error: + self._flush_error = error + raise + self._buffers[measurement] = [] + self._record_counts[measurement] = 0 self._last_flush_time = time.monotonic() async def _flush_all_unlocked(self) -> None: """Flush all measurements. Must hold lock.""" + self._raise_if_failed() measurements = list(self._buffers.keys()) for measurement in measurements: if self._buffers[measurement]: @@ -176,12 +200,13 @@ async def close(self) -> None: if self._closed: return - # Cancel the timer task before the final flush so it can't race. + self._closing = True + # Wake an idle timer, but let any in-flight write finish. Shielding + # also prevents cancellation of close's caller from cancelling I/O. + self._stop_event.set() task = self._flush_task if task is not None: - task.cancel() - with contextlib.suppress(asyncio.CancelledError): - await task + await asyncio.shield(task) self._flush_task = None async with self._lock: diff --git a/src/arc_client/ingestion/buffered.py b/src/arc_client/ingestion/buffered.py index d666fbd..9f23843 100644 --- a/src/arc_client/ingestion/buffered.py +++ b/src/arc_client/ingestion/buffered.py @@ -58,6 +58,8 @@ def __init__( self._last_flush_time = time.monotonic() self._lock = threading.Lock() self._closed = False + self._closing = False + self._flush_error: BaseException | None = None # Background timer that flushes on flush_interval even when no writes # arrive. Started on __enter__ and torn down on close(). @@ -99,6 +101,9 @@ def write(self, record: dict[str, Any]) -> None: columns[key] = [value] with self._lock: + self._raise_if_failed() + if self._closed or self._closing: + raise RuntimeError("Buffered writer is closed") self._buffers[measurement].append(columns) self._record_counts[measurement] += 1 @@ -127,6 +132,9 @@ def write_columnar( num_records = len(next(iter(columns.values()))) with self._lock: + self._raise_if_failed() + if self._closed or self._closing: + raise RuntimeError("Buffered writer is closed") self._buffers[measurement].append(columns) self._record_counts[measurement] += num_records @@ -139,7 +147,12 @@ def write_columnar( self._flush_all() def flush(self) -> None: - """Manually flush all buffered data.""" + """Flush all buffered data, waiting for any in-flight write. + + A failed write leaves its data pending and makes the buffer terminal: + subsequent flush/close calls raise that failure without retrying data + whose server outcome may be unknown. Producers wait while I/O runs. + """ with self._lock: self._flush_all() @@ -155,15 +168,25 @@ def _start_flush_thread(self) -> None: def _flush_loop(self) -> None: """Periodically flush measurements whose data has aged past the interval.""" - while not self._stop_event.wait(self._flush_interval): + try: + while not self._stop_event.wait(self._flush_interval): + with self._lock: + if self._closed or self._stop_event.is_set(): + return + if ( + self._buffers + and time.monotonic() - self._last_flush_time >= self._flush_interval + ): + self._flush_all() + except Exception as error: + # Keep the error observable on foreground operations; never retry + # an uncertain write automatically from the background thread. with self._lock: - if self._closed: - return - if ( - self._buffers - and time.monotonic() - self._last_flush_time >= self._flush_interval - ): - self._flush_all() + self._flush_error = error + + def _raise_if_failed(self) -> None: + if self._flush_error is not None: + raise self._flush_error def _flush_measurement(self, measurement: str) -> None: """Flush a single measurement's buffer. Must hold lock.""" @@ -173,21 +196,21 @@ def _flush_measurement(self, measurement: str) -> None: # Merge all columnar batches into one merged = self._merge_columnar(self._buffers[measurement]) - # Clear buffer - self._buffers[measurement] = [] - self._record_counts[measurement] = 0 - - # Write to Arc (release lock during I/O) - self._lock.release() + # Keep ownership and data until the write is acknowledged. Holding + # the lock across I/O serializes timer, manual, and close flushes. try: self._client.write_columnar(measurement, merged) - finally: - self._lock.acquire() + except BaseException as error: + self._flush_error = error + raise + self._buffers[measurement] = [] + self._record_counts[measurement] = 0 self._last_flush_time = time.monotonic() def _flush_all(self) -> None: """Flush all measurements. Must hold lock.""" + self._raise_if_failed() measurements = list(self._buffers.keys()) for measurement in measurements: if self._buffers[measurement]: @@ -229,11 +252,12 @@ def close(self) -> None: if self._closed: return - # Stop the timer thread first so it can't race the final flush. + self._closing = True + # Stop new timer work and wait for any active write to finish. self._stop_event.set() thread = self._flush_thread if thread is not None: - thread.join(timeout=self._flush_interval + 1.0) + thread.join() self._flush_thread = None with self._lock: diff --git a/tests/unit/http/test_error_diagnostics.py b/tests/unit/http/test_error_diagnostics.py new file mode 100644 index 0000000..0d5a7f8 --- /dev/null +++ b/tests/unit/http/test_error_diagnostics.py @@ -0,0 +1,38 @@ +"""Structured HTTP diagnostics survive existing public exception wrappers.""" + +import inspect +from contextlib import AsyncExitStack + +import pytest + +from arc_client import ArcClient, AsyncArcClient +from arc_client.exceptions import ArcError + + +@pytest.mark.parametrize("async_mode", [False, True], ids=["sync", "async"]) +@pytest.mark.parametrize("operation", ["query", "write", "cq-execute"]) +async def test_wrapper_retains_response_body(httpx_mock, async_mode, operation): + body = {"error": "Permission denied: read required", "detail": "server diagnostic"} + httpx_mock.add_response(status_code=403, json=body) + async with AsyncExitStack() as stack: + if async_mode: + client = await stack.enter_async_context(AsyncArcClient(max_retries=0)) + else: + client = stack.enter_context(ArcClient(max_retries=0)) + with pytest.raises(ArcError) as failure: + if operation == "query": + value = client.query.query("SELECT 1") + elif operation == "write": + value = client.write.write_line_protocol("cpu value=1i") + else: + value = client.continuous_queries.execute(1) + if inspect.isawaitable(value): + await value + assert failure.value.response_body == body + + +def test_error_body_handles_cause_cycle(): + first, second = ArcError("first"), ArcError("second") + first.__cause__ = second + second.__cause__ = first + assert first.response_body is None diff --git a/tests/unit/http/test_retry_safety.py b/tests/unit/http/test_retry_safety.py new file mode 100644 index 0000000..27bde19 --- /dev/null +++ b/tests/unit/http/test_retry_safety.py @@ -0,0 +1,67 @@ +"""Mutation outcomes must not be hidden by automatically replaying writes.""" + +import inspect +from contextlib import AsyncExitStack + +import httpx +import pytest + +from arc_client import ArcClient, AsyncArcClient +from arc_client.exceptions import ArcError + + +@pytest.mark.parametrize("async_mode", [False, True], ids=["sync", "async"]) +@pytest.mark.parametrize( + "error", [httpx.ReadTimeout, httpx.WriteTimeout, httpx.RemoteProtocolError] +) +async def test_uncertain_write_is_sent_once(httpx_mock, async_mode, error): + calls = [] + + def failed_write(request): + calls.append(request.content) + raise error("unknown write outcome") + + httpx_mock.add_callback(failed_write, is_reusable=True) + async with AsyncExitStack() as stack: + if async_mode: + client = await stack.enter_async_context(AsyncArcClient(max_retries=3, retry_backoff=0)) + else: + client = stack.enter_context(ArcClient(max_retries=3, retry_backoff=0)) + with pytest.raises(ArcError): + result = client.write.write_line_protocol("cpu,host=a value=1i") + if inspect.isawaitable(result): + await result + assert len(calls) == 1 + + +@pytest.mark.parametrize("async_mode", [False, True], ids=["sync", "async"]) +@pytest.mark.parametrize("status", [429, 503]) +async def test_write_http_failure_is_not_replayed(httpx_mock, async_mode, status): + httpx_mock.add_response( + status_code=status, json={"error": "write outcome unknown"}, is_reusable=True + ) + async with AsyncExitStack() as stack: + if async_mode: + client = await stack.enter_async_context(AsyncArcClient(max_retries=3, retry_backoff=0)) + else: + client = stack.enter_context(ArcClient(max_retries=3, retry_backoff=0)) + with pytest.raises(ArcError): + result = client.write.write_line_protocol("cpu,host=a value=1i") + if inspect.isawaitable(result): + await result + assert len(httpx_mock.get_requests()) == 1 + + +@pytest.mark.parametrize("async_mode", [False, True], ids=["sync", "async"]) +async def test_presend_connection_failure_can_retry_write(httpx_mock, async_mode): + httpx_mock.add_exception(httpx.ConnectError("connection not established")) + httpx_mock.add_response(json={"success": True}) + async with AsyncExitStack() as stack: + if async_mode: + client = await stack.enter_async_context(AsyncArcClient(max_retries=1, retry_backoff=0)) + else: + client = stack.enter_context(ArcClient(max_retries=1, retry_backoff=0)) + result = client.write.write_line_protocol("cpu,host=a value=1i") + if inspect.isawaitable(result): + await result + assert len(httpx_mock.get_requests()) == 2 diff --git a/tests/unit/test_buffered_lifecycle.py b/tests/unit/test_buffered_lifecycle.py new file mode 100644 index 0000000..b83e31a --- /dev/null +++ b/tests/unit/test_buffered_lifecycle.py @@ -0,0 +1,286 @@ +"""Deterministic ownership, shutdown, and failure checks for buffered writes.""" + +import asyncio +import threading + +import pytest + +from arc_client.ingestion.async_buffered import AsyncBufferedWriter +from arc_client.ingestion.buffered import BufferedWriter + + +class SyncWriter: + def __init__(self, error=None): + self.entered = threading.Event() + self.release = threading.Event() + self.calls = [] + self.committed = [] + self.error = error + + def write_columnar(self, measurement, columns): + self.calls.append((measurement, columns)) + self.entered.set() + assert self.release.wait(3), "test did not release write" + if self.error is not None: + raise self.error + self.committed.append((measurement, columns)) + + +class AsyncWriter: + def __init__(self, error=None): + self.entered = asyncio.Event() + self.release = asyncio.Event() + self.calls = [] + self.committed = [] + self.error = error + self.cancelled = False + + async def write_columnar(self, measurement, columns): + self.calls.append((measurement, columns)) + self.entered.set() + try: + await asyncio.wait_for(self.release.wait(), 3) + except asyncio.CancelledError: + self.cancelled = True + raise + if self.error is not None: + raise self.error + self.committed.append((measurement, columns)) + + +async def next_turn(): + """Let already scheduled tasks reach their next suspension without sleeping.""" + reached = asyncio.Event() + asyncio.get_running_loop().call_soon(reached.set) + await reached.wait() + + +async def start_async_timer(buffer): + buffer._last_flush_time = 0 + buffer._flush_interval = 0.001 + await buffer.__aenter__() + + +def start_sync_timer(buffer): + buffer._last_flush_time = 0 + buffer._flush_interval = 0.001 + buffer.__enter__() + + +async def test_async_close_waits_for_active_timer_write_without_cancelling(): + writer = AsyncWriter() + buffer = AsyncBufferedWriter(writer, flush_interval=60) + await buffer.write_columnar("cpu", {"time": [1], "value": [2]}) + await start_async_timer(buffer) + await asyncio.wait_for(writer.entered.wait(), 3) + closing = asyncio.create_task(buffer.close()) + await next_turn() + writer.release.set() + await asyncio.wait_for(closing, 3) + assert writer.cancelled is False + assert len(writer.committed) == len(writer.calls) == 1 + assert buffer.pending_count == 0 + + +async def test_async_manual_flush_and_close_serialize_with_timer(): + writer = AsyncWriter() + buffer = AsyncBufferedWriter(writer, flush_interval=60) + await buffer.write_columnar("cpu", {"time": [1]}) + await start_async_timer(buffer) + await asyncio.wait_for(writer.entered.wait(), 3) + flushing = asyncio.create_task(buffer.flush()) + closing = asyncio.create_task(buffer.close()) + await next_turn() + writer.release.set() + await asyncio.wait_for(asyncio.gather(flushing, closing), 3) + assert writer.cancelled is False + assert len(writer.committed) == len(writer.calls) == 1 + + +async def test_async_concurrent_closes_do_not_duplicate_inflight_write(): + writer = AsyncWriter() + buffer = AsyncBufferedWriter(writer, flush_interval=60) + await buffer.write_columnar("cpu", {"time": [1]}) + await start_async_timer(buffer) + await asyncio.wait_for(writer.entered.wait(), 3) + first = asyncio.create_task(buffer.close()) + second = asyncio.create_task(buffer.close()) + await next_turn() + writer.release.set() + await asyncio.wait_for(asyncio.gather(first, second), 3) + assert len(writer.committed) == len(writer.calls) == 1 + + +async def test_async_cancelling_close_propagates_without_cancelling_timer_io(): + writer = AsyncWriter() + buffer = AsyncBufferedWriter(writer, flush_interval=60) + await buffer.write_columnar("cpu", {"time": [1]}) + await start_async_timer(buffer) + await asyncio.wait_for(writer.entered.wait(), 3) + closing = asyncio.create_task(buffer.close()) + await next_turn() + closing.cancel() + with pytest.raises(asyncio.CancelledError): + await closing + assert writer.cancelled is False + assert buffer.pending_count == 1 + writer.release.set() + await asyncio.wait_for(buffer.close(), 3) + assert len(writer.committed) == len(writer.calls) == 1 + + +async def test_async_cancelling_manual_flush_retains_data_and_propagates(): + writer = AsyncWriter() + buffer = AsyncBufferedWriter(writer, flush_interval=60) + await buffer.write_columnar("cpu", {"time": [1]}) + flushing = asyncio.create_task(buffer.flush()) + await asyncio.wait_for(writer.entered.wait(), 3) + flushing.cancel() + with pytest.raises(asyncio.CancelledError): + await flushing + assert buffer.pending_count == 1 + with pytest.raises(asyncio.CancelledError): + await buffer.close() + assert len(writer.calls) == 1 + + +def test_sync_close_and_manual_flush_wait_for_active_timer(monkeypatch): + writer = SyncWriter() + buffer = BufferedWriter(writer, flush_interval=60) + buffer.write_columnar("cpu", {"time": [1]}) + start_sync_timer(buffer) + assert writer.entered.wait(3) + joining = threading.Event() + timer_join = buffer._flush_thread.join + + def observed_join(*args, **kwargs): + joining.set() + timer_join(*args, **kwargs) + + monkeypatch.setattr(buffer._flush_thread, "join", observed_join) + closing = threading.Thread(target=buffer.close) + closing.start() + assert joining.wait(3) + flushing = threading.Thread(target=buffer.flush) + flushing.start() + writer.release.set() + closing.join(3) + flushing.join(3) + assert not closing.is_alive() and not flushing.is_alive() + assert len(writer.committed) == len(writer.calls) == 1 + assert buffer.pending_count == 0 + + +async def test_async_idle_close_wakes_timer_immediately(): + buffer = AsyncBufferedWriter(AsyncWriter(), flush_interval=3600) + await buffer.__aenter__() + task = buffer._flush_task + await next_turn() + await asyncio.wait_for(buffer.close(), 1) + assert task.done() + + +@pytest.mark.parametrize("background", [False, True]) +async def test_async_failure_retains_all_unsent_data_and_close_surfaces_error(background): + error = RuntimeError("write outcome unknown") + writer = AsyncWriter(error) + buffer = AsyncBufferedWriter(writer, flush_interval=60) + await buffer.write_columnar("cpu", {"time": [1]}) + await buffer.write_columnar("mem", {"time": [2]}) + writer.release.set() + if background: + await start_async_timer(buffer) + await asyncio.wait_for(writer.entered.wait(), 3) + # Observe task completion, allowing the implementation to record the error. + await asyncio.gather(buffer._flush_task, return_exceptions=True) + else: + with pytest.raises(RuntimeError, match="outcome unknown"): + await buffer.flush() + assert buffer.pending_count == 2 + for operation in (buffer.close, buffer.flush, buffer.close): + with pytest.raises(RuntimeError) as caught: + await operation() + assert caught.value is error + assert len(writer.calls) == 1 + + +@pytest.mark.parametrize("background", [False, True]) +def test_sync_failure_retains_all_unsent_data_and_close_surfaces_error(background): + error = RuntimeError("write outcome unknown") + writer = SyncWriter(error) + buffer = BufferedWriter(writer, flush_interval=60) + buffer.write_columnar("cpu", {"time": [1]}) + buffer.write_columnar("mem", {"time": [2]}) + writer.release.set() + if background: + start_sync_timer(buffer) + assert writer.entered.wait(3) + buffer._flush_thread.join(3) + assert not buffer._flush_thread.is_alive() + else: + with pytest.raises(RuntimeError, match="outcome unknown"): + buffer.flush() + assert buffer.pending_count == 2 + for operation in (buffer.close, buffer.flush, buffer.close): + with pytest.raises(RuntimeError) as caught: + operation() + assert caught.value is error + assert len(writer.calls) == 1 + + +async def test_async_successful_batches_clear_but_failed_batch_remains(): + class FailSecond: + calls = [] + + async def write_columnar(self, measurement, columns): + self.calls.append(measurement) + if measurement == "mem": + raise RuntimeError("second batch failed") + + writer = FailSecond() + buffer = AsyncBufferedWriter(writer, flush_interval=60) + await buffer.write_columnar("cpu", {"time": [1]}) + await buffer.write_columnar("mem", {"time": [2]}) + with pytest.raises(RuntimeError): + await buffer.flush() + assert writer.calls == ["cpu", "mem"] + assert buffer.pending_count == 1 + + +def test_sync_successful_batches_clear_but_failed_batch_remains(): + class FailSecond: + calls = [] + + def write_columnar(self, measurement, columns): + self.calls.append(measurement) + if measurement == "mem": + raise RuntimeError("second batch failed") + + writer = FailSecond() + buffer = BufferedWriter(writer, flush_interval=60) + buffer.write_columnar("cpu", {"time": [1]}) + buffer.write_columnar("mem", {"time": [2]}) + with pytest.raises(RuntimeError): + buffer.flush() + assert writer.calls == ["cpu", "mem"] + assert buffer.pending_count == 1 + + +async def test_async_sparse_padding_survives_lifecycle_changes(): + writer = AsyncWriter() + writer.release.set() + buffer = AsyncBufferedWriter(writer, flush_interval=60) + await buffer.write_columnar("cpu", {"time": [1], "a": [10]}) + await buffer.write_columnar("cpu", {"time": [2], "b": [20]}) + await buffer.close() + assert writer.committed == [("cpu", {"time": [1, 2], "a": [10, None], "b": [None, 20]})] + + +def test_sync_sparse_padding_survives_lifecycle_changes(): + writer = SyncWriter() + writer.release.set() + buffer = BufferedWriter(writer, flush_interval=60) + buffer.write_columnar("cpu", {"time": [1], "a": [10]}) + buffer.write_columnar("cpu", {"time": [2], "b": [20]}) + buffer.close() + assert writer.committed == [("cpu", {"time": [1, 2], "a": [10, None], "b": [None, 20]})]