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