fix(http): prevent mutation replay and retain buffered batches

.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

CHANGELOG.md

@@ -5,7 +5,7 @@ All notable changes to `arc-tsdb-client` are documented here.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
-## [0.2.0] - 2026-07-03
+## [0.2.0] - Unreleased
 
 A large feature-parity release bringing the SDK up to date with the Arc server,
 plus bug fixes and the first real HTTP-layer test coverage.
@@ -28,9 +28,10 @@ plus bug fixes and the first real HTTP-layer test coverage.
   `mqtt`.
 - **zstd write compression** (now the default; gzip and none still supported).
   Compression is configurable via a `Compression` enum, a string, or a bool.
-- **Automatic retries** with exponential backoff + jitter for connection errors,
-  429 (honoring `Retry-After`), and 503. Configurable via `max_retries` /
-  `retry_backoff`.
+- **Automatic retries** with exponential backoff + jitter for read-only
+  connection/transport failures, 429 (honoring `Retry-After`) and 503. Mutations
+  retry only connection establishment; data DELETE never retries. Configure
+  via `max_retries` / `retry_backoff`.
 - New exceptions: `ArcNotSupportedError` (501) and `ArcLicenseError` (403 for a
   missing enterprise-license feature; subclass of `ArcAuthenticationError`).
 - `ArcPermissionError` (403: the token lacks the permission the endpoint
@@ -65,6 +66,9 @@ plus bug fixes and the first real HTTP-layer test coverage.
 
 ### Fixed
 
+- Buffered close waits for active writes, retains unacknowledged batches and
+  surfaces failures without replay. Typed errors retain structured diagnostics.
+
 - Polars datetime → columnar conversion used a fragile private attribute that
   raised at runtime; it now casts via the public `polars` API and normalizes the
   series' `time_unit` to microseconds.

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,

src/arc_client/exceptions.py

@@ -2,13 +2,29 @@
 
 from __future__ import annotations
 
-from typing import Optional
+from typing import Any, Optional
 
 
 class ArcError(Exception):
-    """Base exception for arc-client."""
-
-    pass
+    """Base exception; HTTP failures retain structured diagnostics when available."""
+
+    _response_body: Optional[dict[str, Any]] = None
+
+    @property
+    def response_body(self) -> Optional[dict[str, Any]]:
+        """Retain diagnostics through existing higher-level exception wrappers."""
+        error: Optional[BaseException] = self
+        seen: set[int] = set()
+        while isinstance(error, ArcError) and id(error) not in seen:
+            seen.add(id(error))
+            if error._response_body is not None:
+                return error._response_body
+            error = error.__cause__
+        return None
+
+    @response_body.setter
+    def response_body(self, value: Optional[dict[str, Any]]) -> None:
+        self._response_body = value
 
 
 class ArcConnectionError(ArcError):

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)

src/arc_client/http/base.py

@@ -13,6 +13,7 @@
 from arc_client.exceptions import (
     ArcAuthenticationError,
     ArcConnectionError,
+    ArcError,
     ArcLicenseError,
     ArcNotFoundError,
     ArcNotSupportedError,
@@ -100,6 +101,19 @@ def _is_permission_data_unavailable(message: object) -> bool:
 
 
 def handle_response_error(response: httpx.Response) -> None:
+    """Raise the existing typed error, retaining the server's diagnostic object."""
+    try:
+        _raise_response_error(response)
+    except ArcError as error:
+        try:
+            body = response.json()
+        except ValueError:
+            body = None
+        error.response_body = body if isinstance(body, dict) else None
+        raise
+
+
+def _raise_response_error(response: httpx.Response) -> None:
     """Handle HTTP error responses and raise appropriate exceptions.
 
     Raises:
@@ -185,7 +199,7 @@ def handle_connection_error(error: Exception, url: str) -> None:
     raise ArcConnectionError(f"Failed to connect to {url}: {error}") from error
 
 
-# Transport errors that are safe to retry (connection refused, DNS, timeouts).
+# Transient transport errors eligible for retry on read-only requests.
 RETRYABLE_TRANSPORT_ERRORS = (
     httpx.ConnectError,
     httpx.ConnectTimeout,

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