From 1f301028c499cf6eb821950b9d475b073f558476 Mon Sep 17 00:00:00 2001 From: lecodev-26 Date: Sun, 1 Nov 2026 10:12:05 +0100 Subject: [PATCH 1/4] feat(schedulers): expose scheduler status and controls --- CHANGELOG.md | 7 +- README.md | 29 ++++- src/arc_client/async_client.py | 10 ++ src/arc_client/client.py | 10 ++ src/arc_client/http/base.py | 8 +- src/arc_client/management/__init__.py | 4 + src/arc_client/management/async_schedulers.py | 69 +++++++++++ src/arc_client/management/schedulers.py | 69 +++++++++++ tests/unit/http/test_schedulers.py | 117 ++++++++++++++++++ 9 files changed, 320 insertions(+), 3 deletions(-) create mode 100644 src/arc_client/management/async_schedulers.py create mode 100644 src/arc_client/management/schedulers.py create mode 100644 tests/unit/http/test_schedulers.py diff --git a/CHANGELOG.md b/CHANGELOG.md index bebc788..1c675eb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -79,6 +79,11 @@ plus bug fixes and the first real HTTP-layer test coverage. `ARC_CLIENT_SEND_INSTALLATION_ID=0` or `send_installation_id=False`; pin the ID in containers with `ARC_CLIENT_INSTALLATION_ID`. See README, "Privacy". +- **Scheduler management** (`client.schedulers`, sync + async): inspect CQ and retention + scheduler status, reload active continuous-query schedules, and manually trigger + retention. The status route accepts authenticated tokens; control routes preserve + Arc administrator and enterprise-license gates. Retention triggers require + `confirm=True` because they may permanently delete data. Supported by Arc 26.09.3 and later. - Arrow completion metadata in the existing `pyarrow.Table` schema for sync and async queries: policy cap and execution time are retained; failed, malformed, conflicting or unobservable announced completion metadata is rejected. @@ -108,7 +113,7 @@ plus bug fixes and the first real HTTP-layer test coverage. 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 +- New exceptions: `ArcNotSupportedError` (501) and `ArcLicenseError` (403/503 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 diff --git a/README.md b/README.md index eca88fe..9a7f483 100644 --- a/README.md +++ b/README.md @@ -64,7 +64,7 @@ with ArcClient(host="localhost", token="your-token") as client: - **Databases**: List, create, delete databases and list measurements - **Management**: Retention policies, continuous queries, delete operations - **Enterprise**: RBAC, query governance, audit logs, query management, cluster, - compaction, storage tiering, backup/restore, and MQTT ingestion + compaction, scheduler controls, storage tiering, backup/restore, and MQTT ingestion - **Authentication**: Token management (create, rotate, revoke) ## Data Ingestion @@ -401,6 +401,28 @@ print(f"Would delete {result.deleted_count} rows") result = client.retention.execute(policy.id, dry_run=False, confirm=True) ``` +### Scheduler inspection and controls + +Scheduler status is available to authenticated tokens. Reloading continuous-query +schedules and manually triggering retention require an administrator token and +the corresponding enterprise scheduler feature. + +```python +status = client.schedulers.status() +client.schedulers.reload_continuous_queries() + +# Runs all active retention policies immediately and may permanently delete data. +client.schedulers.trigger_retention(confirm=True) +``` + +The retention operation refuses to send a request unless `confirm=True` is +supplied. The async client exposes the same methods: + +```python +async with AsyncArcClient(host="localhost", token="your-token") as client: + status = await client.schedulers.status() +``` + ### Continuous Queries Aggregate and downsample data automatically: @@ -513,6 +535,11 @@ client.compaction.trigger(database="metrics", measurement="cpu") client.compaction.cycles(limit=5) client.compaction.too_wide() +# Scheduler inspection and explicit operator controls +client.schedulers.status() # requires an authenticated token +client.schedulers.reload_continuous_queries() # administrator + licensed feature +# Retention may permanently delete data; explicit confirmation is required. +client.schedulers.trigger_retention(confirm=True) client.tiering.set_policy("metrics", hot_only=True, hot_max_age_days=7) client.backup.create(databases=["metrics"]) # scoped backup diff --git a/src/arc_client/async_client.py b/src/arc_client/async_client.py index cc8c1d9..f7b5713 100644 --- a/src/arc_client/async_client.py +++ b/src/arc_client/async_client.py @@ -101,6 +101,7 @@ def __init__( self._edge_sync: Any = None self._diagnostics: Any = None self._reconciliation: Any = None + self._schedulers: Any = None def _get_http(self) -> AsyncHTTPClient: """Get or create the HTTP client.""" @@ -259,6 +260,15 @@ def backup(self) -> Any: self._backup = AsyncBackupClient(self._get_http(), self._config) return self._backup + @property + def schedulers(self) -> Any: + """Get the async scheduler inspection and control client.""" + if self._schedulers is None: + from arc_client.management.async_schedulers import AsyncSchedulerClient + + self._schedulers = AsyncSchedulerClient(self._get_http(), self._config) + return self._schedulers + @property def mqtt(self) -> Any: """Get the async MQTT ingestion management client (enterprise).""" diff --git a/src/arc_client/client.py b/src/arc_client/client.py index 5a5a1ae..ef9d551 100644 --- a/src/arc_client/client.py +++ b/src/arc_client/client.py @@ -101,6 +101,7 @@ def __init__( self._edge_sync: Any = None self._diagnostics: Any = None self._reconciliation: Any = None + self._schedulers: Any = None def _get_http(self) -> SyncHTTPClient: """Get or create the HTTP client.""" @@ -257,6 +258,15 @@ def backup(self) -> Any: self._backup = BackupClient(self._get_http(), self._config) return self._backup + @property + def schedulers(self) -> Any: + """Get the scheduler inspection and control client.""" + if self._schedulers is None: + from arc_client.management.schedulers import SchedulerClient + + self._schedulers = SchedulerClient(self._get_http(), self._config) + return self._schedulers + @property def mqtt(self) -> Any: """Get the MQTT ingestion management client (enterprise).""" diff --git a/src/arc_client/http/base.py b/src/arc_client/http/base.py index 899a8f8..1922939 100644 --- a/src/arc_client/http/base.py +++ b/src/arc_client/http/base.py @@ -132,7 +132,7 @@ def _raise_response_error(response: httpx.Response) -> None: Raises: ArcAuthenticationError: HTTP 401 — the token is missing, invalid or expired. Re-authenticate. - ArcLicenseError: HTTP 403 naming an enterprise license. + ArcLicenseError: HTTP 403 or 503 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``. @@ -202,6 +202,12 @@ def _raise_response_error(response: httpx.Response) -> None: raise ArcRateLimitError(f"Rate limit exceeded: {message}") elif status_code == 501: raise ArcNotSupportedError(f"Operation not supported by server: {message}") + elif ( + status_code == 503 + and isinstance(message, str) + and "license required" in message.lower() + ): + raise ArcLicenseError(f"Enterprise license required: {message}") elif (refusal := ingest_refusal(response)) is not None: reason, retry_safe = refusal raise ArcIngestRefusedError( diff --git a/src/arc_client/management/__init__.py b/src/arc_client/management/__init__.py index f155a87..1e60757 100644 --- a/src/arc_client/management/__init__.py +++ b/src/arc_client/management/__init__.py @@ -20,6 +20,7 @@ from arc_client.management.async_rbac import AsyncRBACClient from arc_client.management.async_reconciliation import AsyncReconciliationClient from arc_client.management.async_retention import AsyncRetentionClient +from arc_client.management.async_schedulers import AsyncSchedulerClient from arc_client.management.async_tiering import AsyncTieringClient from arc_client.management.audit import AuditClient from arc_client.management.backup import BackupClient @@ -36,6 +37,7 @@ from arc_client.management.rbac import RBACClient from arc_client.management.reconciliation import ReconciliationClient from arc_client.management.retention import RetentionClient +from arc_client.management.schedulers import SchedulerClient from arc_client.management.tiering import TieringClient __all__ = [ @@ -46,6 +48,8 @@ "AsyncDeleteClient", "RetentionClient", "AsyncRetentionClient", + "SchedulerClient", + "AsyncSchedulerClient", "DatabaseClient", "AsyncDatabaseClient", # Enterprise diff --git a/src/arc_client/management/async_schedulers.py b/src/arc_client/management/async_schedulers.py new file mode 100644 index 0000000..7084d1b --- /dev/null +++ b/src/arc_client/management/async_schedulers.py @@ -0,0 +1,69 @@ +"""Asynchronous scheduler inspection and control for Arc. + +Wraps ``/api/v1/schedulers``. Status is available to authenticated tokens; +control operations require an administrator token and the corresponding +enterprise scheduler feature. Arc reports a missing scheduler license as +``ArcLicenseError`` even when the route responds with HTTP 503. +""" + +from __future__ import annotations + +from typing import Any + +from arc_client.config import ClientConfig +from arc_client.exceptions import ArcError, ArcValidationError +from arc_client.http.async_http import AsyncHTTPClient + + +class AsyncSchedulerClient: + """Inspect scheduler status and perform explicit scheduler controls. + + Methods return raw JSON values because scheduler status and counters are + server-defined operator diagnostics, not a stable typed response schema. + """ + + def __init__(self, http: AsyncHTTPClient, config: ClientConfig) -> None: + self._http = http + self._config = config + + async def _json(self, method: str, path: str, **kwargs: Any) -> Any: + try: + response = await getattr(self._http, method)(path, **kwargs) + return response.json() + except ArcError: + raise + except Exception as e: + raise ArcError(f"Scheduler request failed ({method.upper()} {path}): {e}") from e + + async def status(self) -> Any: + """Return the status of the CQ and retention schedulers. + + Available with any authenticated token. Returns raw JSON with each + scheduler's enabled state and, when unavailable, its reason. + """ + return await self._json("get", "/api/v1/schedulers/") + + async def reload_continuous_queries(self) -> Any: + """Reload active CQ schedules (admin; raises ArcLicenseError without license). + + Returns the raw JSON response from Arc. + """ + return await self._json("post", "/api/v1/schedulers/cq/reload") + + async def trigger_retention(self, confirm: bool = False) -> Any: + """Run all active retention policies immediately (administrator only). + + This can delete data according to the configured retention policies. + Pass ``confirm=True`` to explicitly authorize the immediate run. + Returns the raw JSON response from Arc. + + Raises: + ArcValidationError: If ``confirm`` is not True. + ArcLicenseError: If the scheduler feature is not licensed. + """ + if not confirm: + raise ArcValidationError( + "Refusing to trigger retention without confirm=True. " + "This may permanently delete data." + ) + return await self._json("post", "/api/v1/schedulers/retention/trigger") diff --git a/src/arc_client/management/schedulers.py b/src/arc_client/management/schedulers.py new file mode 100644 index 0000000..23c8fdb --- /dev/null +++ b/src/arc_client/management/schedulers.py @@ -0,0 +1,69 @@ +"""Synchronous scheduler inspection and control for Arc. + +Wraps ``/api/v1/schedulers``. Status is available to authenticated tokens; +control operations require an administrator token and the corresponding +enterprise scheduler feature. Arc reports a missing scheduler license as +``ArcLicenseError`` even when the route responds with HTTP 503. +""" + +from __future__ import annotations + +from typing import Any + +from arc_client.config import ClientConfig +from arc_client.exceptions import ArcError, ArcValidationError +from arc_client.http.sync_http import SyncHTTPClient + + +class SchedulerClient: + """Inspect scheduler status and perform explicit scheduler controls. + + Methods return raw JSON values because scheduler status and counters are + server-defined operator diagnostics, not a stable typed response schema. + """ + + def __init__(self, http: SyncHTTPClient, config: ClientConfig) -> None: + self._http = http + self._config = config + + def _json(self, method: str, path: str, **kwargs: Any) -> Any: + try: + response = getattr(self._http, method)(path, **kwargs) + return response.json() + except ArcError: + raise + except Exception as e: + raise ArcError(f"Scheduler request failed ({method.upper()} {path}): {e}") from e + + def status(self) -> Any: + """Return the status of the CQ and retention schedulers. + + Available with any authenticated token. Returns raw JSON with each + scheduler's enabled state and, when unavailable, its reason. + """ + return self._json("get", "/api/v1/schedulers/") + + def reload_continuous_queries(self) -> Any: + """Reload active CQ schedules (admin; raises ArcLicenseError without license). + + Returns the raw JSON response from Arc. + """ + return self._json("post", "/api/v1/schedulers/cq/reload") + + def trigger_retention(self, confirm: bool = False) -> Any: + """Run all active retention policies immediately (administrator only). + + This can delete data according to the configured retention policies. + Pass ``confirm=True`` to explicitly authorize the immediate run. + Returns the raw JSON response from Arc. + + Raises: + ArcValidationError: If ``confirm`` is not True. + ArcLicenseError: If the scheduler feature is not licensed. + """ + if not confirm: + raise ArcValidationError( + "Refusing to trigger retention without confirm=True. " + "This may permanently delete data." + ) + return self._json("post", "/api/v1/schedulers/retention/trigger") diff --git a/tests/unit/http/test_schedulers.py b/tests/unit/http/test_schedulers.py new file mode 100644 index 0000000..b693981 --- /dev/null +++ b/tests/unit/http/test_schedulers.py @@ -0,0 +1,117 @@ +"""HTTP contract tests for scheduler inspection and explicit controls.""" + +from __future__ import annotations + +import pytest +from pytest_httpx import HTTPXMock + +from arc_client import ArcClient, AsyncArcClient +from arc_client.exceptions import ( + ArcLicenseError, + ArcPermissionError, + ArcServerError, + ArcValidationError, +) + + +@pytest.mark.parametrize("async_mode", [False, True]) +async def test_scheduler_status(httpx_mock: HTTPXMock, async_mode: bool) -> None: + expected = { + "cq_scheduler": {"enabled": True, "job_count": 2}, + "retention_scheduler": {"enabled": False, "reason": "not configured"}, + } + httpx_mock.add_response(json=expected) + + if async_mode: + async with AsyncArcClient(host="h", token="t", max_retries=0) as client: + result = await client.schedulers.status() + else: + with ArcClient(host="h", token="t", max_retries=0) as client: + result = client.schedulers.status() + + assert result == expected + request = httpx_mock.get_requests()[0] + assert request.method == "GET" + assert request.url.path == "/api/v1/schedulers/" + + +@pytest.mark.parametrize("async_mode", [False, True]) +async def test_scheduler_controls(httpx_mock: HTTPXMock, async_mode: bool) -> None: + httpx_mock.add_response(json={"message": "CQ scheduler reloaded", "job_count": 3}) + httpx_mock.add_response(json={"message": "Retention triggered successfully"}) + + if async_mode: + async with AsyncArcClient(host="h", token="t", max_retries=0) as client: + reload_result = await client.schedulers.reload_continuous_queries() + retention_result = await client.schedulers.trigger_retention(confirm=True) + else: + with ArcClient(host="h", token="t", max_retries=0) as client: + reload_result = client.schedulers.reload_continuous_queries() + retention_result = client.schedulers.trigger_retention(confirm=True) + + assert reload_result["job_count"] == 3 + assert retention_result["message"] == "Retention triggered successfully" + requests = httpx_mock.get_requests() + assert [(request.method, request.url.path) for request in requests] == [ + ("POST", "/api/v1/schedulers/cq/reload"), + ("POST", "/api/v1/schedulers/retention/trigger"), + ] + + +@pytest.mark.parametrize("async_mode", [False, True]) +async def test_retention_trigger_requires_explicit_confirmation( + httpx_mock: HTTPXMock, async_mode: bool +) -> None: + if async_mode: + async with AsyncArcClient(host="h", token="t", max_retries=0) as client, pytest.raises( + ArcValidationError, match="confirm=True" + ): + await client.schedulers.trigger_retention() + else: + with ArcClient(host="h", token="t", max_retries=0) as client, pytest.raises( + ArcValidationError, match="confirm=True" + ): + client.schedulers.trigger_retention() + + assert httpx_mock.get_requests() == [] + + +@pytest.mark.parametrize("async_mode", [False, True]) +async def test_scheduler_controls_preserve_permission_and_server_errors( + httpx_mock: HTTPXMock, async_mode: bool +) -> None: + httpx_mock.add_response(status_code=403, json={"error": "Admin permission required"}) + httpx_mock.add_response(status_code=503, json={"error": "Retention scheduler is not running"}) + + if async_mode: + async with AsyncArcClient(host="h", token="t", max_retries=0) as client: + with pytest.raises(ArcPermissionError): + await client.schedulers.reload_continuous_queries() + with pytest.raises(ArcServerError) as error: + await client.schedulers.trigger_retention(confirm=True) + else: + with ArcClient(host="h", token="t", max_retries=0) as client: + with pytest.raises(ArcPermissionError): + client.schedulers.reload_continuous_queries() + with pytest.raises(ArcServerError) as error: + client.schedulers.trigger_retention(confirm=True) + + assert error.value.status_code == 503 + + +@pytest.mark.parametrize("async_mode", [False, True]) +async def test_scheduler_license_required_503_maps_to_license_error( + httpx_mock: HTTPXMock, async_mode: bool +) -> None: + httpx_mock.add_response( + status_code=503, json={"error": "CQ scheduler is not running - enterprise license required"} + ) + + if async_mode: + async with AsyncArcClient(host="h", token="t", max_retries=0) as client: + with pytest.raises(ArcLicenseError, match="Enterprise license required"): + await client.schedulers.reload_continuous_queries() + else: + with ArcClient(host="h", token="t", max_retries=0) as client: + with pytest.raises(ArcLicenseError, match="Enterprise license required"): + client.schedulers.reload_continuous_queries() From 5461967000f99156c2a92c72d33a81ad6dddd19d Mon Sep 17 00:00:00 2001 From: lecodev-26 Date: Mon, 2 Nov 2026 22:24:24 +0100 Subject: [PATCH 2/4] fix(schedulers): preserve 503 and extend retention timeout --- CHANGELOG.md | 2 +- src/arc_client/http/base.py | 8 +---- src/arc_client/management/async_schedulers.py | 29 ++++++++++++++----- src/arc_client/management/schedulers.py | 29 ++++++++++++++----- tests/unit/http/test_schedulers.py | 16 +++++----- 5 files changed, 52 insertions(+), 32 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1c675eb..ff47b2e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -113,7 +113,7 @@ plus bug fixes and the first real HTTP-layer test coverage. 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/503 for a +- 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 diff --git a/src/arc_client/http/base.py b/src/arc_client/http/base.py index 1922939..899a8f8 100644 --- a/src/arc_client/http/base.py +++ b/src/arc_client/http/base.py @@ -132,7 +132,7 @@ def _raise_response_error(response: httpx.Response) -> None: Raises: ArcAuthenticationError: HTTP 401 — the token is missing, invalid or expired. Re-authenticate. - ArcLicenseError: HTTP 403 or 503 naming an enterprise license. + 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``. @@ -202,12 +202,6 @@ def _raise_response_error(response: httpx.Response) -> None: raise ArcRateLimitError(f"Rate limit exceeded: {message}") elif status_code == 501: raise ArcNotSupportedError(f"Operation not supported by server: {message}") - elif ( - status_code == 503 - and isinstance(message, str) - and "license required" in message.lower() - ): - raise ArcLicenseError(f"Enterprise license required: {message}") elif (refusal := ingest_refusal(response)) is not None: reason, retry_safe = refusal raise ArcIngestRefusedError( diff --git a/src/arc_client/management/async_schedulers.py b/src/arc_client/management/async_schedulers.py index 7084d1b..150c45e 100644 --- a/src/arc_client/management/async_schedulers.py +++ b/src/arc_client/management/async_schedulers.py @@ -2,8 +2,9 @@ Wraps ``/api/v1/schedulers``. Status is available to authenticated tokens; control operations require an administrator token and the corresponding -enterprise scheduler feature. Arc reports a missing scheduler license as -``ArcLicenseError`` even when the route responds with HTTP 503. +enterprise scheduler feature. A missing enterprise feature is reported as +HTTP 403; HTTP 503 means the scheduler is not running, and ``status()`` +provides the reason. """ from __future__ import annotations @@ -13,6 +14,7 @@ from arc_client.config import ClientConfig from arc_client.exceptions import ArcError, ArcValidationError from arc_client.http.async_http import AsyncHTTPClient +from arc_client.management.delete import DELETE_TIMEOUT_FLOOR class AsyncSchedulerClient: @@ -44,9 +46,11 @@ async def status(self) -> Any: return await self._json("get", "/api/v1/schedulers/") async def reload_continuous_queries(self) -> Any: - """Reload active CQ schedules (admin; raises ArcLicenseError without license). + """Reload active CQ schedules (administrator and licensed feature required). - Returns the raw JSON response from Arc. + Returns the raw JSON response from Arc. A missing enterprise feature + raises ``ArcLicenseError`` (HTTP 403). HTTP 503 means the scheduler is + not running; call ``status()`` to inspect the reason. """ return await self._json("post", "/api/v1/schedulers/cq/reload") @@ -54,16 +58,25 @@ async def trigger_retention(self, confirm: bool = False) -> Any: """Run all active retention policies immediately (administrator only). This can delete data according to the configured retention policies. - Pass ``confirm=True`` to explicitly authorize the immediate run. - Returns the raw JSON response from Arc. + Pass ``confirm=True`` to explicitly authorize the immediate run. The + request allows at least 90 seconds because Arc runs every active policy + synchronously. In a cluster, send it to the primary writer; another node + returns HTTP 500. A running retention cycle also returns HTTP 500. + + Returns the raw JSON response from Arc. A missing enterprise feature + raises ``ArcLicenseError`` (HTTP 403). HTTP 503 means the scheduler is + not running; call ``status()`` to inspect the reason. Raises: ArcValidationError: If ``confirm`` is not True. - ArcLicenseError: If the scheduler feature is not licensed. """ if not confirm: raise ArcValidationError( "Refusing to trigger retention without confirm=True. " "This may permanently delete data." ) - return await self._json("post", "/api/v1/schedulers/retention/trigger") + return await self._json( + "post", + "/api/v1/schedulers/retention/trigger", + timeout=max(self._config.timeout, DELETE_TIMEOUT_FLOOR), + ) diff --git a/src/arc_client/management/schedulers.py b/src/arc_client/management/schedulers.py index 23c8fdb..96ff2dd 100644 --- a/src/arc_client/management/schedulers.py +++ b/src/arc_client/management/schedulers.py @@ -2,8 +2,9 @@ Wraps ``/api/v1/schedulers``. Status is available to authenticated tokens; control operations require an administrator token and the corresponding -enterprise scheduler feature. Arc reports a missing scheduler license as -``ArcLicenseError`` even when the route responds with HTTP 503. +enterprise scheduler feature. A missing enterprise feature is reported as +HTTP 403; HTTP 503 means the scheduler is not running, and ``status()`` +provides the reason. """ from __future__ import annotations @@ -13,6 +14,7 @@ from arc_client.config import ClientConfig from arc_client.exceptions import ArcError, ArcValidationError from arc_client.http.sync_http import SyncHTTPClient +from arc_client.management.delete import DELETE_TIMEOUT_FLOOR class SchedulerClient: @@ -44,9 +46,11 @@ def status(self) -> Any: return self._json("get", "/api/v1/schedulers/") def reload_continuous_queries(self) -> Any: - """Reload active CQ schedules (admin; raises ArcLicenseError without license). + """Reload active CQ schedules (administrator and licensed feature required). - Returns the raw JSON response from Arc. + Returns the raw JSON response from Arc. A missing enterprise feature + raises ``ArcLicenseError`` (HTTP 403). HTTP 503 means the scheduler is + not running; call ``status()`` to inspect the reason. """ return self._json("post", "/api/v1/schedulers/cq/reload") @@ -54,16 +58,25 @@ def trigger_retention(self, confirm: bool = False) -> Any: """Run all active retention policies immediately (administrator only). This can delete data according to the configured retention policies. - Pass ``confirm=True`` to explicitly authorize the immediate run. - Returns the raw JSON response from Arc. + Pass ``confirm=True`` to explicitly authorize the immediate run. The + request allows at least 90 seconds because Arc runs every active policy + synchronously. In a cluster, send it to the primary writer; another node + returns HTTP 500. A running retention cycle also returns HTTP 500. + + Returns the raw JSON response from Arc. A missing enterprise feature + raises ``ArcLicenseError`` (HTTP 403). HTTP 503 means the scheduler is + not running; call ``status()`` to inspect the reason. Raises: ArcValidationError: If ``confirm`` is not True. - ArcLicenseError: If the scheduler feature is not licensed. """ if not confirm: raise ArcValidationError( "Refusing to trigger retention without confirm=True. " "This may permanently delete data." ) - return self._json("post", "/api/v1/schedulers/retention/trigger") + return self._json( + "post", + "/api/v1/schedulers/retention/trigger", + timeout=max(self._config.timeout, DELETE_TIMEOUT_FLOOR), + ) diff --git a/tests/unit/http/test_schedulers.py b/tests/unit/http/test_schedulers.py index b693981..7211433 100644 --- a/tests/unit/http/test_schedulers.py +++ b/tests/unit/http/test_schedulers.py @@ -7,7 +7,6 @@ from arc_client import ArcClient, AsyncArcClient from arc_client.exceptions import ( - ArcLicenseError, ArcPermissionError, ArcServerError, ArcValidationError, @@ -63,10 +62,9 @@ async def test_retention_trigger_requires_explicit_confirmation( httpx_mock: HTTPXMock, async_mode: bool ) -> None: if async_mode: - async with AsyncArcClient(host="h", token="t", max_retries=0) as client, pytest.raises( - ArcValidationError, match="confirm=True" - ): - await client.schedulers.trigger_retention() + async with AsyncArcClient(host="h", token="t", max_retries=0) as client: + with pytest.raises(ArcValidationError, match="confirm=True"): + await client.schedulers.trigger_retention() else: with ArcClient(host="h", token="t", max_retries=0) as client, pytest.raises( ArcValidationError, match="confirm=True" @@ -100,7 +98,7 @@ async def test_scheduler_controls_preserve_permission_and_server_errors( @pytest.mark.parametrize("async_mode", [False, True]) -async def test_scheduler_license_required_503_maps_to_license_error( +async def test_scheduler_license_required_503_remains_server_error( httpx_mock: HTTPXMock, async_mode: bool ) -> None: httpx_mock.add_response( @@ -109,9 +107,11 @@ async def test_scheduler_license_required_503_maps_to_license_error( if async_mode: async with AsyncArcClient(host="h", token="t", max_retries=0) as client: - with pytest.raises(ArcLicenseError, match="Enterprise license required"): + with pytest.raises(ArcServerError) as error: await client.schedulers.reload_continuous_queries() else: with ArcClient(host="h", token="t", max_retries=0) as client: - with pytest.raises(ArcLicenseError, match="Enterprise license required"): + with pytest.raises(ArcServerError) as error: client.schedulers.reload_continuous_queries() + + assert error.value.status_code == 503 From c0b451b379d7848642b3ac87caf8fe177e2d93ad Mon Sep 17 00:00:00 2001 From: lecodev-26 Date: Mon, 2 Nov 2026 22:26:18 +0100 Subject: [PATCH 3/4] style(schedulers): combine sync error test contexts --- tests/unit/http/test_schedulers.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/tests/unit/http/test_schedulers.py b/tests/unit/http/test_schedulers.py index 7211433..adea34a 100644 --- a/tests/unit/http/test_schedulers.py +++ b/tests/unit/http/test_schedulers.py @@ -110,8 +110,9 @@ async def test_scheduler_license_required_503_remains_server_error( with pytest.raises(ArcServerError) as error: await client.schedulers.reload_continuous_queries() else: - with ArcClient(host="h", token="t", max_retries=0) as client: - with pytest.raises(ArcServerError) as error: - client.schedulers.reload_continuous_queries() + with ArcClient(host="h", token="t", max_retries=0) as client, pytest.raises( + ArcServerError + ) as error: + client.schedulers.reload_continuous_queries() assert error.value.status_code == 503 From 33f8a97a691690ea7591255c29f7fbceb1c75f8e Mon Sep 17 00:00:00 2001 From: Ignacio Van Droogenbroeck <64545348+xe-nvdk@users.noreply.github.com> Date: Wed, 4 Nov 2026 10:24:46 -0600 Subject: [PATCH 4/4] fix(schedulers): follow Arc 27.01.1 scheduler refusals (arc#1445) Arc#1445 answers a scheduler control refused for the license with 403, a scheduler off in config with 503 naming the setting, and a retention trigger on a non-primary node or during a running cycle with 409. Only the "Enterprise license required" form of the 403 matched the generic license detection; the tier and refused-reload forms name the license without that phrase and surfaced as ArcPermissionError. The scheduler clients now raise ArcLicenseError for a 403 that names the license, keeping the response body. Docstrings and the changelog give the 403/503/409 meanings, and the tests use Arc's actual messages instead of the pre-#1445 503 "license required" wording. Also fixes the ruff format failure in tests/unit/http/test_schedulers.py. Claude-Session: https://claude.ai/code/session_01VMc4wABiN8pxeMccxkQRpw --- CHANGELOG.md | 5 ++ src/arc_client/management/async_schedulers.py | 48 +++++++--- src/arc_client/management/schedulers.py | 48 +++++++--- tests/unit/http/test_schedulers.py | 88 ++++++++++++++++--- 4 files changed, 150 insertions(+), 39 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ff47b2e..0277f72 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -84,6 +84,11 @@ plus bug fixes and the first real HTTP-layer test coverage. retention. The status route accepts authenticated tokens; control routes preserve Arc administrator and enterprise-license gates. Retention triggers require `confirm=True` because they may permanently delete data. Supported by Arc 26.09.3 and later. + On Arc 27.01.1 and later, a control refused because the license is missing, expired + or of a tier without the scheduler raises `ArcLicenseError` (HTTP 403); a scheduler + off in config is `ArcServerError` with `status_code == 503`, and a retention trigger + on a node that is not the primary writer, or while a cycle runs, is `ArcServerError` + with `status_code == 409`. - Arrow completion metadata in the existing `pyarrow.Table` schema for sync and async queries: policy cap and execution time are retained; failed, malformed, conflicting or unobservable announced completion metadata is rejected. diff --git a/src/arc_client/management/async_schedulers.py b/src/arc_client/management/async_schedulers.py index 150c45e..8025b83 100644 --- a/src/arc_client/management/async_schedulers.py +++ b/src/arc_client/management/async_schedulers.py @@ -2,9 +2,10 @@ Wraps ``/api/v1/schedulers``. Status is available to authenticated tokens; control operations require an administrator token and the corresponding -enterprise scheduler feature. A missing enterprise feature is reported as -HTTP 403; HTTP 503 means the scheduler is not running, and ``status()`` -provides the reason. +enterprise scheduler feature. Since Arc 27.01.1 a missing, expired or +insufficient license is HTTP 403, raised as ``ArcLicenseError``; HTTP 503 +means the scheduler is off in config or failed to start, and ``status()`` +gives the reason. Earlier Arc versions answer 503 in both cases. """ from __future__ import annotations @@ -12,7 +13,12 @@ from typing import Any from arc_client.config import ClientConfig -from arc_client.exceptions import ArcError, ArcValidationError +from arc_client.exceptions import ( + ArcError, + ArcLicenseError, + ArcPermissionError, + ArcValidationError, +) from arc_client.http.async_http import AsyncHTTPClient from arc_client.management.delete import DELETE_TIMEOUT_FLOOR @@ -32,6 +38,17 @@ async def _json(self, method: str, path: str, **kwargs: Any) -> Any: try: response = await getattr(self._http, method)(path, **kwargs) return response.json() + except ArcPermissionError as e: + # Arc refuses a scheduler control with 403 when the license is + # missing, expired or of a tier without the scheduler. Only the + # "Enterprise license required" form matches the generic 403 + # license detection; the tier and refused-reload forms name the + # license without that phrase. + message = str(e) + if "license" in message.lower(): + detail = message.split("Permission denied: ", 1)[-1] + raise ArcLicenseError(f"Enterprise license required: {detail}") from e + raise except ArcError: raise except Exception as e: @@ -48,9 +65,10 @@ async def status(self) -> Any: async def reload_continuous_queries(self) -> Any: """Reload active CQ schedules (administrator and licensed feature required). - Returns the raw JSON response from Arc. A missing enterprise feature - raises ``ArcLicenseError`` (HTTP 403). HTTP 503 means the scheduler is - not running; call ``status()`` to inspect the reason. + Returns the raw JSON response from Arc. A license that is missing, + expired or does not include the scheduler raises ``ArcLicenseError`` + (HTTP 403). HTTP 503 means the scheduler is off in config or failed to + start; call ``status()`` for the reason. """ return await self._json("post", "/api/v1/schedulers/cq/reload") @@ -60,12 +78,16 @@ async def trigger_retention(self, confirm: bool = False) -> Any: This can delete data according to the configured retention policies. Pass ``confirm=True`` to explicitly authorize the immediate run. The request allows at least 90 seconds because Arc runs every active policy - synchronously. In a cluster, send it to the primary writer; another node - returns HTTP 500. A running retention cycle also returns HTTP 500. - - Returns the raw JSON response from Arc. A missing enterprise feature - raises ``ArcLicenseError`` (HTTP 403). HTTP 503 means the scheduler is - not running; call ``status()`` to inspect the reason. + synchronously. In a cluster, send it to the primary writer: another node + refuses it, as Arc does while a retention cycle is already running. + Since Arc 27.01.1 both refusals are HTTP 409, raised as + ``ArcServerError`` with ``status_code == 409``; earlier versions + answer HTTP 500. + + Returns the raw JSON response from Arc. A license that is missing, + expired or does not include the scheduler raises ``ArcLicenseError`` + (HTTP 403). HTTP 503 means the scheduler is off in config or failed to + start; call ``status()`` for the reason. Raises: ArcValidationError: If ``confirm`` is not True. diff --git a/src/arc_client/management/schedulers.py b/src/arc_client/management/schedulers.py index 96ff2dd..4b0d7e7 100644 --- a/src/arc_client/management/schedulers.py +++ b/src/arc_client/management/schedulers.py @@ -2,9 +2,10 @@ Wraps ``/api/v1/schedulers``. Status is available to authenticated tokens; control operations require an administrator token and the corresponding -enterprise scheduler feature. A missing enterprise feature is reported as -HTTP 403; HTTP 503 means the scheduler is not running, and ``status()`` -provides the reason. +enterprise scheduler feature. Since Arc 27.01.1 a missing, expired or +insufficient license is HTTP 403, raised as ``ArcLicenseError``; HTTP 503 +means the scheduler is off in config or failed to start, and ``status()`` +gives the reason. Earlier Arc versions answer 503 in both cases. """ from __future__ import annotations @@ -12,7 +13,12 @@ from typing import Any from arc_client.config import ClientConfig -from arc_client.exceptions import ArcError, ArcValidationError +from arc_client.exceptions import ( + ArcError, + ArcLicenseError, + ArcPermissionError, + ArcValidationError, +) from arc_client.http.sync_http import SyncHTTPClient from arc_client.management.delete import DELETE_TIMEOUT_FLOOR @@ -32,6 +38,17 @@ def _json(self, method: str, path: str, **kwargs: Any) -> Any: try: response = getattr(self._http, method)(path, **kwargs) return response.json() + except ArcPermissionError as e: + # Arc refuses a scheduler control with 403 when the license is + # missing, expired or of a tier without the scheduler. Only the + # "Enterprise license required" form matches the generic 403 + # license detection; the tier and refused-reload forms name the + # license without that phrase. + message = str(e) + if "license" in message.lower(): + detail = message.split("Permission denied: ", 1)[-1] + raise ArcLicenseError(f"Enterprise license required: {detail}") from e + raise except ArcError: raise except Exception as e: @@ -48,9 +65,10 @@ def status(self) -> Any: def reload_continuous_queries(self) -> Any: """Reload active CQ schedules (administrator and licensed feature required). - Returns the raw JSON response from Arc. A missing enterprise feature - raises ``ArcLicenseError`` (HTTP 403). HTTP 503 means the scheduler is - not running; call ``status()`` to inspect the reason. + Returns the raw JSON response from Arc. A license that is missing, + expired or does not include the scheduler raises ``ArcLicenseError`` + (HTTP 403). HTTP 503 means the scheduler is off in config or failed to + start; call ``status()`` for the reason. """ return self._json("post", "/api/v1/schedulers/cq/reload") @@ -60,12 +78,16 @@ def trigger_retention(self, confirm: bool = False) -> Any: This can delete data according to the configured retention policies. Pass ``confirm=True`` to explicitly authorize the immediate run. The request allows at least 90 seconds because Arc runs every active policy - synchronously. In a cluster, send it to the primary writer; another node - returns HTTP 500. A running retention cycle also returns HTTP 500. - - Returns the raw JSON response from Arc. A missing enterprise feature - raises ``ArcLicenseError`` (HTTP 403). HTTP 503 means the scheduler is - not running; call ``status()`` to inspect the reason. + synchronously. In a cluster, send it to the primary writer: another node + refuses it, as Arc does while a retention cycle is already running. + Since Arc 27.01.1 both refusals are HTTP 409, raised as + ``ArcServerError`` with ``status_code == 409``; earlier versions + answer HTTP 500. + + Returns the raw JSON response from Arc. A license that is missing, + expired or does not include the scheduler raises ``ArcLicenseError`` + (HTTP 403). HTTP 503 means the scheduler is off in config or failed to + start; call ``status()`` for the reason. Raises: ArcValidationError: If ``confirm`` is not True. diff --git a/tests/unit/http/test_schedulers.py b/tests/unit/http/test_schedulers.py index adea34a..c1356a5 100644 --- a/tests/unit/http/test_schedulers.py +++ b/tests/unit/http/test_schedulers.py @@ -2,11 +2,14 @@ from __future__ import annotations +from typing import Any + import pytest from pytest_httpx import HTTPXMock from arc_client import ArcClient, AsyncArcClient from arc_client.exceptions import ( + ArcLicenseError, ArcPermissionError, ArcServerError, ArcValidationError, @@ -66,8 +69,9 @@ async def test_retention_trigger_requires_explicit_confirmation( with pytest.raises(ArcValidationError, match="confirm=True"): await client.schedulers.trigger_retention() else: - with ArcClient(host="h", token="t", max_retries=0) as client, pytest.raises( - ArcValidationError, match="confirm=True" + with ( + ArcClient(host="h", token="t", max_retries=0) as client, + pytest.raises(ArcValidationError, match="confirm=True"), ): client.schedulers.trigger_retention() @@ -97,22 +101,80 @@ async def test_scheduler_controls_preserve_permission_and_server_errors( assert error.value.status_code == 503 +# The refusals Arc 27.01.1 answers on the control routes (arc#1433, arc#1445). +_CQ = "/api/v1/schedulers/cq/reload" +_RETENTION = "/api/v1/schedulers/retention/trigger" + + @pytest.mark.parametrize("async_mode", [False, True]) -async def test_scheduler_license_required_503_remains_server_error( - httpx_mock: HTTPXMock, async_mode: bool +@pytest.mark.parametrize( + ("path", "message"), + [ + (_CQ, "CQ scheduler is not running: Enterprise license required"), + (_CQ, "CQ scheduler is not running: License tier does not include CQ scheduler feature"), + (_CQ, "CQ scheduler reload refused: cq: the license does not include the CQ scheduler"), + ( + _RETENTION, + "Retention trigger refused: retention: the license does not include the " + "retention scheduler", + ), + ], +) +async def test_scheduler_license_refusals_raise_license_error( + httpx_mock: HTTPXMock, async_mode: bool, path: str, message: str ) -> None: - httpx_mock.add_response( - status_code=503, json={"error": "CQ scheduler is not running - enterprise license required"} - ) + httpx_mock.add_response(status_code=403, json={"error": message}) + + if async_mode: + async with AsyncArcClient(host="h", token="t", max_retries=0) as client: + with pytest.raises(ArcLicenseError) as error: + await _control(client.schedulers, path) + else: + with ( + ArcClient(host="h", token="t", max_retries=0) as client, + pytest.raises(ArcLicenseError) as error, + ): + _control(client.schedulers, path) + + assert message in str(error.value) + assert error.value.response_body == {"error": message} + + +@pytest.mark.parametrize("async_mode", [False, True]) +@pytest.mark.parametrize( + ("path", "status", "message"), + [ + ( + _CQ, + 503, + "CQ scheduler is not running: CQ feature not enabled " + "(continuous_query.enabled=false), or the scheduler failed to start " + "(see the startup log)", + ), + (_RETENTION, 409, "Retention trigger refused: retention: node role is not primary writer"), + (_RETENTION, 409, "Retention trigger refused: retention: a cycle is already running"), + ], +) +async def test_scheduler_not_running_and_refusals_are_server_errors( + httpx_mock: HTTPXMock, async_mode: bool, path: str, status: int, message: str +) -> None: + httpx_mock.add_response(status_code=status, json={"error": message}) if async_mode: async with AsyncArcClient(host="h", token="t", max_retries=0) as client: with pytest.raises(ArcServerError) as error: - await client.schedulers.reload_continuous_queries() + await _control(client.schedulers, path) else: - with ArcClient(host="h", token="t", max_retries=0) as client, pytest.raises( - ArcServerError - ) as error: - client.schedulers.reload_continuous_queries() + with ( + ArcClient(host="h", token="t", max_retries=0) as client, + pytest.raises(ArcServerError) as error, + ): + _control(client.schedulers, path) - assert error.value.status_code == 503 + assert error.value.status_code == status + + +def _control(schedulers: Any, path: str) -> Any: + if path == _CQ: + return schedulers.reload_continuous_queries() + return schedulers.trigger_retention(confirm=True)