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 1eb68f4..4baaefc 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,11 +28,21 @@ 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 + 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,9 +56,22 @@ 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 +- 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. + - 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/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..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 @@ -17,13 +18,17 @@ ArcLicenseError, ArcNotFoundError, ArcNotSupportedError, + ArcPermissionDataUnavailableError, + ArcPermissionError, ArcQueryError, ArcRateLimitError, + ArcScopedAccessError, ArcServerError, ArcValidationError, ) __all__ = [ + "CLEAR", # Clients "ArcClient", "AsyncArcClient", @@ -42,6 +47,9 @@ "ArcServerError", "ArcNotSupportedError", "ArcLicenseError", + "ArcPermissionError", + "ArcPermissionDataUnavailableError", + "ArcScopedAccessError", # Version "__version__", ] 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/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 a7b36bf..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.""" + """Base exception; HTTP failures retain structured diagnostics when available.""" - pass + _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): @@ -79,3 +95,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/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 d92e359..1cb8a3f 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 @@ -12,10 +13,14 @@ from arc_client.exceptions import ( ArcAuthenticationError, ArcConnectionError, + ArcError, ArcLicenseError, ArcNotFoundError, ArcNotSupportedError, + ArcPermissionDataUnavailableError, + ArcPermissionError, ArcRateLimitError, + ArcScopedAccessError, ArcServerError, ) @@ -38,8 +43,96 @@ 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.""" + """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: + 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 +151,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: @@ -76,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/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_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/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/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/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_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 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_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) 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]})]