fix: findings from live testing 0.2.0 against Arc main (single node, edge sync, cluster)

CHANGELOG.md

@@ -158,7 +158,9 @@ plus bug fixes and the first real HTTP-layer test coverage.
   themselves.
 - `create_token(permissions=[])` and `update_token(permissions=[])` send the
   empty list, creating an RBAC-only token, instead of silently producing a
-  read/write token.
+  read/write token. On Arc, such a token gets access only where RBAC grants
+  are consulted (the `/api/v1/databases` routes); SQL query routes check the
+  coarse `read` permission, so it cannot run queries.
 - `HealthResponse` models Arc's real `/health` (`time`, `uptime_sec`,
   `storage`, `license`).
 - `dataframe_to_columnar(tag_columns=...)` is no longer a no-op.

src/arc_client/http/async_http.py

@@ -139,6 +139,7 @@ async def post(
         data: Optional[Any] = None,
         idempotent: bool = False,
         retry_if: Optional[RetryPredicate] = None,
+        timeout: Optional[float] = None,
     ) -> httpx.Response:
         """Make a POST request.
 
@@ -155,6 +156,8 @@ async def post(
             extra["files"] = files
         if data is not None:
             extra["data"] = data
+        if timeout is not None:
+            extra["timeout"] = timeout
         return await self._request(
             "POST",
             path,
@@ -196,9 +199,11 @@ async def delete(
         path: str,
         params: Optional[dict[str, Any]] = None,
         headers: Optional[dict[str, str]] = None,
+        timeout: Optional[float] = None,
     ) -> httpx.Response:
         """Make a DELETE request."""
-        return await self._request("DELETE", path, headers=headers, params=params)
+        extra: dict[str, Any] = {"timeout": timeout} if timeout is not None else {}
+        return await self._request("DELETE", path, headers=headers, params=params, **extra)
 
     async def __aenter__(self) -> AsyncHTTPClient:
         return self

src/arc_client/http/base.py

@@ -216,8 +216,18 @@ def _raise_response_error(response: httpx.Response) -> None:
 
 
 def handle_connection_error(error: Exception, url: str) -> None:
-    """Handle connection errors and raise ArcConnectionError."""
-    raise ArcConnectionError(f"Failed to connect to {url}: {error}") from error
+    """Handle connection errors and raise ArcConnectionError.
+
+    Only a failure to connect means the request never reached Arc. A read
+    timeout or a connection lost mid-response happens after the request was
+    sent, so the operation may well have run on the server.
+    """
+    if isinstance(error, (httpx.ConnectError, httpx.ConnectTimeout, httpx.PoolTimeout)):
+        raise ArcConnectionError(f"Failed to connect to {url}: {error}") from error
+    raise ArcConnectionError(
+        f"No complete response from {url} ({type(error).__name__}: {error}); the request "
+        "was sent and may have been applied by Arc. Check before retrying a mutation."
+    ) from error
 
 
 # Transient transport errors eligible for retry on read-only requests.
@@ -282,6 +292,15 @@ def is_retryable_write_refusal(response: httpx.Response) -> bool:
     return refusal is not None and refusal[1]
 
 
+def retry_unavailable_only(response: httpx.Response) -> bool:
+    """Status-retry predicate for governed reads: retry 503, never 429.
+
+    Arc's query governance answers 429 (rate limit or quota) with no
+    Retry-After, and it does not clear within a few seconds of backoff.
+    """
+    return response.status_code == 503
+
+
 def never_retry_status(response: httpx.Response) -> bool:
     """Status-retry predicate that disables status retries for one request."""
     return False

src/arc_client/http/sync_http.py

@@ -137,6 +137,7 @@ def post(
         data: Optional[Any] = None,
         idempotent: bool = False,
         retry_if: Optional[RetryPredicate] = None,
+        timeout: Optional[float] = None,
     ) -> httpx.Response:
         """Make a POST request.
 
@@ -153,6 +154,8 @@ def post(
             extra["files"] = files
         if data is not None:
             extra["data"] = data
+        if timeout is not None:
+            extra["timeout"] = timeout
         return self._request(
             "POST",
             path,
@@ -194,9 +197,11 @@ def delete(
         path: str,
         params: Optional[dict[str, Any]] = None,
         headers: Optional[dict[str, str]] = None,
+        timeout: Optional[float] = None,
     ) -> httpx.Response:
         """Make a DELETE request."""
-        return self._request("DELETE", path, headers=headers, params=params)
+        extra: dict[str, Any] = {"timeout": timeout} if timeout is not None else {}
+        return self._request("DELETE", path, headers=headers, params=params, **extra)
 
     def __enter__(self) -> SyncHTTPClient:
         return self

src/arc_client/management/async_cluster.py

@@ -18,6 +18,7 @@
 from __future__ import annotations
 
 from typing import Any, Optional, Union
+from urllib.parse import quote
 
 from arc_client.config import ClientConfig
 from arc_client.exceptions import ArcError, ArcValidationError
@@ -112,8 +113,20 @@ async def delete_file(
         return await self._json("delete", "/api/v1/cluster/files", params=params)
 
     async def remove_node(self, node_id: str) -> Any:
-        """Remove a node from the cluster via Raft (admin)."""
-        return await self._json("delete", f"/api/v1/cluster/nodes/{node_id}")
+        """Remove a node from the cluster via Raft (admin).
+
+        Raises:
+            ArcError: If Arc did not remove it. A node without clustering
+                answers 200 with ``success: false``, which would otherwise
+                read as a successful removal.
+        """
+        if not node_id:
+            raise ArcValidationError("node_id is required")
+        result = await self._json("delete", f"/api/v1/cluster/nodes/{quote(node_id, safe='')}")
+        if isinstance(result, dict) and result.get("success") is False:
+            reason = result.get("error") or result.get("reason") or "clustering is not enabled"
+            raise ArcError(f"Node {node_id!r} was not removed: {reason}")
+        return result
 
     async def demote_writer(self, node_id: str, confirm: bool = False) -> Any:
         """Hand the primary-writer role off ``node_id`` so the cluster elects another (admin).
@@ -138,8 +151,11 @@ async def assign_compactor(self, node_id: str) -> Any:
 
         The target is explicit and nothing is deleted, so no ``confirm``.
         Arc refuses (409) readers, unhealthy nodes, the current holder and
-        non-leader nodes. The response's ``preemption_suppressed_seconds``
-        says how long the choice is protected from automatic preemption.
+        non-leader nodes, and, before it even checks the target, a cluster
+        that does not manage a compactor lease (no node has the compactor
+        role), so an unknown node then gets 409 rather than 404. The
+        response's ``preemption_suppressed_seconds`` says how long the choice
+        is protected from automatic preemption.
         Needs Arc 2026.09.2+.
         """
         if not node_id:

src/arc_client/management/async_continuous_query.py

@@ -229,7 +229,12 @@ async def execute(
         end_time: Union[datetime, str, None] = None,
         dry_run: bool = False,
     ) -> ExecuteCQResponse:
-        """Execute a continuous query manually."""
+        """Execute a continuous query manually.
+
+        Arc refuses (400) an ``end_time`` more than a minute in the future.
+        ``watermark_status`` on the result is ``None`` on standalone nodes
+        and on Arc releases before 2027.01.1.
+        """
         payload: dict[str, Any] = {"dry_run": dry_run}
         if start_time:
             payload["start_time"] = (

src/arc_client/management/async_databases.py

@@ -7,6 +7,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
 from arc_client.models.database import (
     DatabaseInfo,
     DatabaseListResponse,
@@ -185,7 +186,11 @@ async def delete(
             params = {"confirm": "true"}
             if allow_unreachable:
                 params["allow_unreachable"] = "true"
-            response = await self._http.delete(f"/api/v1/databases/{name}", params=params)
+            response = await self._http.delete(
+                f"/api/v1/databases/{name}",
+                params=params,
+                timeout=max(self._config.timeout, DELETE_TIMEOUT_FLOOR),
+            )
             body: dict[str, Any] = response.json()
             return body
         except ArcError:

src/arc_client/management/async_delete.py

@@ -7,6 +7,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
 from arc_client.models.delete import DeleteConfigResponse, DeleteResponse
 
 
@@ -34,6 +35,9 @@ async def delete(
             where: SQL WHERE clause selecting the rows.
             dry_run: Preview only (default). Set False to delete.
             confirm: Required by Arc for any delete that is not a dry run.
+                In a cluster, send it to the primary writer: other nodes
+                refuse a data delete (503 "not primary writer") instead of
+                forwarding it, unlike writes.
             allow_unreachable: Cluster only; relaxes a safety check. Without
                 it, Arc refuses the DELETE (503) when files the cluster
                 manifest lists are not on this node yet (stream takeover), or
@@ -64,7 +68,8 @@ async def delete(
             payload["allow_unreachable"] = True
 
         try:
-            response = await self._http.post("/api/v1/delete", json=payload)
+            timeout = None if dry_run else max(self._config.timeout, DELETE_TIMEOUT_FLOOR)
+            response = await self._http.post("/api/v1/delete", json=payload, timeout=timeout)
             data = response.json()
             return DeleteResponse(**data)
         except ArcError:

src/arc_client/management/async_edge_sync.py

@@ -34,6 +34,7 @@
 from arc_client.config import ClientConfig
 from arc_client.exceptions import ArcError
 from arc_client.http.async_http import AsyncHTTPClient
+from arc_client.http.base import never_retry_status
 from arc_client.management.edge_sync import (
     BUNDLE_IMPORT,
     DELETE_REFUSAL,
@@ -86,14 +87,16 @@ async def get_spoke(self, spoke_id: str) -> Spoke:
             await self._dict("get", f"{SPOKES}/{_segment('spoke_id', spoke_id)}")
         )
 
-    async def register_spoke(self, spoke_id: str, name: str = "") -> SpokeSecret:
+    async def register_spoke(self, spoke_id: str, name: Optional[str] = None) -> SpokeSecret:
         """Register a spoke and get its HMAC secret (hub).
 
         The secret is returned only this once; store it before discarding the
         result. 409 if the ID is already registered (Arc refuses to reissue).
+        ``name`` defaults to ``spoke_id``: Arc requires a non-empty name
+        (internal/edgesync/registry.go).
         """
         _segment("spoke_id", spoke_id)
-        body = {"spoke_id": spoke_id, "name": name}
+        body = {"spoke_id": spoke_id, "name": name or spoke_id}
         return SpokeSecret.model_validate(await self._dict("post", f"{SPOKES}/", json=body))
 
     async def rotate_spoke_secret(self, spoke_id: str, confirm: bool = False) -> SpokeSecret:
@@ -167,9 +170,11 @@ async def spoke_status(self) -> dict[str, Any]:
 
         ``pending``, ``in_flight``, ``exported``, ``synced``, ``failed``,
         ``skipped``, ``pending_bytes`` and, once anything has synced,
-        ``last_synced_at`` / ``seconds_since_last_sync``.
+        ``last_synced_at`` / ``seconds_since_last_sync``. Files are entered
+        in the ledger by a sync or export pass, so ``pending`` stays 0 for
+        new data until one runs.
         """
-        return await self._dict("get", f"{SPOKE_SYNC}/status")
+        return await self._dict("get", f"{SPOKE_SYNC}/status", retry_if=never_retry_status)
 
     async def spoke_ledger(
         self, limit: Optional[int] = None, state: Optional[str] = None
@@ -181,7 +186,12 @@ async def spoke_ledger(
         ``exported``, ``failed`` or ``skipped``. ``limit`` is 1-1000 (server
         default 100).
         """
-        return await self._dict("get", f"{SPOKE_SYNC}/ledger", params=_ledger_params(limit, state))
+        return await self._dict(
+            "get",
+            f"{SPOKE_SYNC}/ledger",
+            params=_ledger_params(limit, state),
+            retry_if=never_retry_status,
+        )
 
     async def run_sync(self) -> dict[str, Any]:
         """Run one network sync pass now and wait for its result (spoke).

src/arc_client/management/async_reconciliation.py

@@ -19,6 +19,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.http.base import never_retry_status
 from arc_client.management.reconciliation import _run_from, _trigger_params
 from arc_client.models.reconciliation import ReconciliationRun
 
@@ -84,4 +85,6 @@ async def get_run(self, run_id: str) -> ReconciliationRun:
         if not run_id:
             raise ArcValidationError("run_id is required")
         path = f"/api/v1/reconciliation/runs/{run_id}"
-        return _run_from(await self._json("get", path), path)
+        # Arc answers a node without reconciliation with a permanent 503
+        # and no Retry-After (reconciliation.go): raise at once.
+        return _run_from(await self._json("get", path, retry_if=never_retry_status), path)

src/arc_client/management/cluster.py

@@ -139,8 +139,20 @@ def delete_file(self, path: str, confirm: bool = False, reason: Optional[str] =
         return self._json("delete", "/api/v1/cluster/files", params=params)
 
     def remove_node(self, node_id: str) -> Any:
-        """Remove a node from the cluster via Raft (admin)."""
-        return self._json("delete", f"/api/v1/cluster/nodes/{node_id}")
+        """Remove a node from the cluster via Raft (admin).
+
+        Raises:
+            ArcError: If Arc did not remove it. A node without clustering
+                answers 200 with ``success: false``, which would otherwise
+                read as a successful removal.
+        """
+        if not node_id:
+            raise ArcValidationError("node_id is required")
+        result = self._json("delete", f"/api/v1/cluster/nodes/{quote(node_id, safe='')}")
+        if isinstance(result, dict) and result.get("success") is False:
+            reason = result.get("error") or result.get("reason") or "clustering is not enabled"
+            raise ArcError(f"Node {node_id!r} was not removed: {reason}")
+        return result
 
     def demote_writer(self, node_id: str, confirm: bool = False) -> Any:
         """Hand the primary-writer role off ``node_id`` so the cluster elects another (admin).
@@ -165,8 +177,11 @@ def assign_compactor(self, node_id: str) -> Any:
 
         The target is explicit and nothing is deleted, so no ``confirm``.
         Arc refuses (409) readers, unhealthy nodes, the current holder and
-        non-leader nodes. The response's ``preemption_suppressed_seconds``
-        says how long the choice is protected from automatic preemption.
+        non-leader nodes, and, before it even checks the target, a cluster
+        that does not manage a compactor lease (no node has the compactor
+        role), so an unknown node then gets 409 rather than 404. The
+        response's ``preemption_suppressed_seconds`` says how long the choice
+        is protected from automatic preemption.
         Needs Arc 2026.09.2+.
         """
         if not node_id:

src/arc_client/management/continuous_query.py

@@ -229,7 +229,12 @@ def execute(
         end_time: Union[datetime, str, None] = None,
         dry_run: bool = False,
     ) -> ExecuteCQResponse:
-        """Execute a continuous query manually."""
+        """Execute a continuous query manually.
+
+        Arc refuses (400) an ``end_time`` more than a minute in the future.
+        ``watermark_status`` on the result is ``None`` on standalone nodes
+        and on Arc releases before 2027.01.1.
+        """
         payload: dict[str, Any] = {"dry_run": dry_run}
         if start_time:
             payload["start_time"] = (

src/arc_client/management/databases.py

@@ -7,6 +7,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
 from arc_client.models.database import (
     DatabaseInfo,
     DatabaseListResponse,
@@ -212,7 +213,11 @@ def delete(
             params = {"confirm": "true"}
             if allow_unreachable:
                 params["allow_unreachable"] = "true"
-            response = self._http.delete(f"/api/v1/databases/{name}", params=params)
+            response = self._http.delete(
+                f"/api/v1/databases/{name}",
+                params=params,
+                timeout=max(self._config.timeout, DELETE_TIMEOUT_FLOOR),
+            )
             body: dict[str, Any] = response.json()
             return body
         except ArcError:

src/arc_client/management/delete.py

@@ -9,6 +9,11 @@
 from arc_client.http.sync_http import SyncHTTPClient
 from arc_client.models.delete import DeleteConfigResponse, DeleteResponse
 
+# Arc may wait up to 30 s inside a delete (internal/api/delete_records.go
+# deleteLocalWait; the database delete's peer flush), so a client timeout near
+# 30 s reports a delete that ran as a failure. Deletes get at least this much.
+DELETE_TIMEOUT_FLOOR = 90.0
+
 
 class DeleteClient:
     """Synchronous client for data deletion operations.
@@ -37,6 +42,9 @@ def delete(
             where: SQL WHERE clause selecting the rows.
             dry_run: Preview only (default). Set False to delete.
             confirm: Required by Arc for any delete that is not a dry run.
+                In a cluster, send it to the primary writer: other nodes
+                refuse a data delete (503 "not primary writer") instead of
+                forwarding it, unlike writes.
             allow_unreachable: Cluster only; relaxes a safety check. Without
                 it, Arc refuses the DELETE (503) when files the cluster
                 manifest lists are not on this node yet (stream takeover), or
@@ -67,7 +75,8 @@ def delete(
             payload["allow_unreachable"] = True
 
         try:
-            response = self._http.post("/api/v1/delete", json=payload)
+            timeout = None if dry_run else max(self._config.timeout, DELETE_TIMEOUT_FLOOR)
+            response = self._http.post("/api/v1/delete", json=payload, timeout=timeout)
             data = response.json()
             return DeleteResponse(**data)
         except ArcError: