From 0b1a8afb7dfe45982f3fe56b2aa16fba57732108 Mon Sep 17 00:00:00 2001 From: Ignacio Van Droogenbroeck <64545348+xe-nvdk@users.noreply.github.com> Date: Mon, 2 Nov 2026 14:39:48 -0600 Subject: [PATCH] feat(write): flush_server_buffers() for Arc's ingest buffer flush POST /api/v1/write/line-protocol/flush (admin) on the sync and async write clients. Arc's line protocol, MessagePack and MQTT paths share one buffer, so it flushes every row the node holds, across databases. Named apart from the client-side BufferedWriter.flush(). The live suites now flush through it instead of private HTTP calls, and the parity suite's docstring lists the features its node must enable. Claude-Session: https://claude.ai/code/session_01VMc4wAG9mUUc3wB2ingB8c --- CHANGELOG.md | 3 + README.md | 9 +++ src/arc_client/ingestion/async_writer.py | 28 +++++++++ src/arc_client/ingestion/writer.py | 28 +++++++++ tests/integration/test_live_cluster.py | 4 +- tests/integration/test_live_parity.py | 18 +++--- tests/integration/test_live_shared.py | 4 +- tests/unit/http/test_flush_server_buffers.py | 65 ++++++++++++++++++++ 8 files changed, 144 insertions(+), 15 deletions(-) create mode 100644 tests/unit/http/test_flush_server_buffers.py diff --git a/CHANGELOG.md b/CHANGELOG.md index f5dc12b..bebc788 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/README.md b/README.md index f284f1e..eca88fe 100644 --- a/README.md +++ b/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: diff --git a/src/arc_client/ingestion/async_writer.py b/src/arc_client/ingestion/async_writer.py index 3df5df3..a576d8e 100644 --- a/src/arc_client/ingestion/async_writer.py +++ b/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, diff --git a/src/arc_client/ingestion/writer.py b/src/arc_client/ingestion/writer.py index 6bf65d2..e0d8359 100644 --- a/src/arc_client/ingestion/writer.py +++ b/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, diff --git a/tests/integration/test_live_cluster.py b/tests/integration/test_live_cluster.py index e25edbf..66da896 100644 --- a/tests/integration/test_live_cluster.py +++ b/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: diff --git a/tests/integration/test_live_parity.py b/tests/integration/test_live_parity.py index 20fc6e1..2118f0b 100644 --- a/tests/integration/test_live_parity.py +++ b/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. ``/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( diff --git a/tests/integration/test_live_shared.py b/tests/integration/test_live_shared.py index 8134693..a6b2c0f 100644 --- a/tests/integration/test_live_shared.py +++ b/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: diff --git a/tests/unit/http/test_flush_server_buffers.py b/tests/unit/http/test_flush_server_buffers.py new file mode 100644 index 0000000..18cecd4 --- /dev/null +++ b/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