feat(write): flush_server_buffers() for Arc ingest buffer flush

CHANGELOG.md

@@ -13,6 +13,9 @@ plus bug fixes and the first real HTTP-layer test coverage.
 ### Added
 
 - **Write path (Arc 2027.01.1 parity).**
+  - `client.write.flush_server_buffers()` (sync + async): flushes the node's
+    ingest buffers to Parquet now (`POST /api/v1/write/line-protocol/flush`,
+    admin). Arc flushes every protocol and database on that node.
   - `ArcPartialWriteError` (`dropped`, `others_written`) for Arc's
     `400 "partial write"`. It is never retried, and buffered writers discard
     that batch and report it once, so a resend cannot duplicate stored rows.

README.md

@@ -126,6 +126,15 @@ with client.write.buffered(batch_size=10000, flush_interval=5.0) as buffer:
     # Auto-flushes on exit or when batch_size/flush_interval reached
 ```
 
+That flush sends rows the client holds. Arc buffers accepted rows in turn and
+writes them to Parquet every `ingest.max_buffer_age_ms` (5 s by default). To
+make them queryable at once, for example in tests, flush the server's buffers
+(admin token; flushes every protocol and database on that node):
+
+```python
+client.write.flush_server_buffers()
+```
+
 ### Line Protocol
 
 For compatibility with InfluxDB tooling:

src/arc_client/ingestion/async_writer.py

@@ -189,6 +189,34 @@ async def tle_stats(self) -> dict[str, Any]:
             raise ArcIngestionError(f"Failed to get TLE stats: {e}") from e
         return tle_stats_body(body)
 
+    async def flush_server_buffers(self) -> dict[str, Any]:
+        """Flush this node's ingest buffers to Parquet now (admin token).
+
+        ``POST /api/v1/write/line-protocol/flush``. Despite the route name,
+        Arc flushes every buffered row on the node it is sent to: line
+        protocol, MessagePack and MQTT share one buffer, across all
+        databases. Rows become visible to queries without waiting for
+        ``ingest.max_buffer_age_ms``. In a cluster it flushes only the node
+        that receives it.
+
+        Not the client-side :meth:`buffered` writer's ``flush()``, which
+        sends rows the client holds; this acts on rows Arc already accepted.
+
+        Returns:
+            Arc's JSON answer, e.g. ``{"status": "success", "message": ...}``.
+
+        Raises:
+            ArcPermissionError: The token is not an admin token.
+        """
+        try:
+            response = await self._http.post("/api/v1/write/line-protocol/flush")
+            body: dict[str, Any] = response.json()
+        except ArcError:
+            raise
+        except Exception as e:
+            raise ArcIngestionError(f"Failed to flush server buffers: {e}") from e
+        return body
+
     def buffered(
         self,
         batch_size: int = 10000,

src/arc_client/ingestion/writer.py

@@ -310,6 +310,34 @@ def tle_stats(self) -> dict[str, Any]:
             raise ArcIngestionError(f"Failed to get TLE stats: {e}") from e
         return tle_stats_body(body)
 
+    def flush_server_buffers(self) -> dict[str, Any]:
+        """Flush this node's ingest buffers to Parquet now (admin token).
+
+        ``POST /api/v1/write/line-protocol/flush``. Despite the route name,
+        Arc flushes every buffered row on the node it is sent to: line
+        protocol, MessagePack and MQTT share one buffer, across all
+        databases. Rows become visible to queries without waiting for
+        ``ingest.max_buffer_age_ms``. In a cluster it flushes only the node
+        that receives it.
+
+        Not the client-side :meth:`buffered` writer's ``flush()``, which
+        sends rows the client holds; this acts on rows Arc already accepted.
+
+        Returns:
+            Arc's JSON answer, e.g. ``{"status": "success", "message": ...}``.
+
+        Raises:
+            ArcPermissionError: The token is not an admin token.
+        """
+        try:
+            response = self._http.post("/api/v1/write/line-protocol/flush")
+            body: dict[str, Any] = response.json()
+        except ArcError:
+            raise
+        except Exception as e:
+            raise ArcIngestionError(f"Failed to flush server buffers: {e}") from e
+        return body
+
     def buffered(
         self,
         batch_size: int = 10000,

tests/integration/test_live_cluster.py

@@ -149,8 +149,8 @@ def _docker_time(value: str) -> float:
 
 
 def flush(client: ArcClient) -> None:
-    """Force the node's ingest buffers to parquet (no SDK wrapper for this route)."""
-    client._get_http().post("/api/v1/write/line-protocol/flush")
+    """Force the node's ingest buffers to parquet."""
+    client.write.flush_server_buffers()
 
 
 def unique(prefix: str) -> str:

tests/integration/test_live_parity.py

@@ -10,6 +10,10 @@
 ``ARC_TEST_LICENSED=1``
     The server has an enterprise license with governance, RBAC, audit,
     query management, backup and compaction enabled. Gates those tests.
+    Several of these are off by default in Arc; start the server with
+    ``ARC_DELETE_ENABLED=true ARC_GOVERNANCE_ENABLED=true
+    ARC_QUERY_MANAGEMENT_ENABLED=true ARC_AUDIT_LOG_ENABLED=true
+    ARC_AUDIT_LOG_INCLUDE_READS=true ARC_RECONCILIATION_ENABLED=true``.
 ``ARC_TEST_STORAGE_DIR``
     The server's local ``storage.local_path`` (e.g. ``<workdir>/data/arc``),
     readable from the test process. Gates the checks that read the Parquet
@@ -140,17 +144,9 @@ def uid(prefix: str) -> str:
 
 
 def server_flush(url: Optional[str] = None, token: Optional[str] = None) -> None:
-    """Flush Arc's ingest buffers (POST /api/v1/write/line-protocol/flush, admin).
-
-    Not an SDK method; used only so the tests need not wait out
-    ``ingest.max_buffer_age_ms``.
-    """
-    response = httpx.post(
-        f"{url or ARC_URL}/api/v1/write/line-protocol/flush",
-        headers={"Authorization": f"Bearer {token or ARC_TOKEN}"},
-        timeout=30,
-    )
-    response.raise_for_status()
+    """Flush Arc's ingest buffers so tests need not wait out ``ingest.max_buffer_age_ms``."""
+    with make_client(url, token) as client:
+        client.write.flush_server_buffers()
 
 
 def wait_rows(

tests/integration/test_live_shared.py

@@ -113,8 +113,8 @@ def docker(*args: str) -> str:
 
 
 def flush(client: ArcClient) -> None:
-    """Force the node's ingest buffers to parquet (no SDK wrapper for this route)."""
-    client._get_http().post("/api/v1/write/line-protocol/flush")
+    """Force the node's ingest buffers to parquet."""
+    client.write.flush_server_buffers()
 
 
 def unique(prefix: str) -> str:

tests/unit/http/test_flush_server_buffers.py

@@ -0,0 +1,65 @@
+"""Server buffer flush wire contract.
+
+Arc source: internal/api/lineprotocol.go (``Flush``, admin-gated).
+"""
+
+from __future__ import annotations
+
+import inspect
+from contextlib import AsyncExitStack
+from typing import Any
+
+import pytest
+from pytest_httpx import HTTPXMock
+
+from arc_client import ArcClient, AsyncArcClient
+from arc_client.exceptions import ArcPermissionError, ArcServerError
+
+URL = "http://localhost:8000/api/v1/write/line-protocol/flush"
+
+
+@pytest.fixture(params=[False, True], ids=["sync", "async"])
+def async_mode(request: Any) -> bool:
+    return bool(request.param)
+
+
+async def flush(async_mode: bool, **kwargs: Any) -> Any:
+    kwargs.setdefault("max_retries", 0)
+    async with AsyncExitStack() as stack:
+        if async_mode:
+            client: Any = await stack.enter_async_context(AsyncArcClient(token="t", **kwargs))
+        else:
+            client = stack.enter_context(ArcClient(token="t", **kwargs))
+        result = client.write.flush_server_buffers()
+        return await result if inspect.isawaitable(result) else result
+
+
+async def test_flush_posts_and_returns_body(httpx_mock: HTTPXMock, async_mode: bool) -> None:
+    body = {"status": "success", "message": "All buffers flushed"}
+    httpx_mock.add_response(method="POST", url=URL, json=body)
+
+    assert await flush(async_mode) == body
+
+    request = httpx_mock.get_request()
+    assert request is not None
+    assert request.headers["Authorization"] == "Bearer t"
+    assert request.content == b""
+
+
+async def test_flush_requires_admin(httpx_mock: HTTPXMock, async_mode: bool) -> None:
+    httpx_mock.add_response(
+        method="POST", url=URL, status_code=403, json={"error": "Permission denied: admin required"}
+    )
+    with pytest.raises(ArcPermissionError):
+        await flush(async_mode)
+
+
+async def test_flush_failure_is_not_retried(httpx_mock: HTTPXMock, async_mode: bool) -> None:
+    httpx_mock.add_response(
+        method="POST", url=URL, status_code=503, json={"error": "Flush failed: disk full"}
+    )
+    with pytest.raises(ArcServerError) as exc:
+        await flush(async_mode, max_retries=3, retry_backoff=0.0)
+    assert exc.value.status_code == 503
+    assert "disk full" in str(exc.value)
+    assert len(httpx_mock.get_requests()) == 1