From d49080b76f5b6590c4740f1a07e5f7672684ac8a Mon Sep 17 00:00:00 2001 From: Ignacio Van Droogenbroeck <64545348+xe-nvdk@users.noreply.github.com> Date: Mon, 2 Nov 2026 10:49:01 -0600 Subject: [PATCH 01/10] test(integration): add opt-in live parity suite for the 0.2.0 surface Exercises every new or changed SDK method against a real Arc built from main: query/profile/query_id/Arrow options, governance 429, precision, tag_columns, partial writes (LP, msgpack rows, buffered), health and readiness, schema, deletes, auth, compaction, backup, CQ, retention, reconciliation, diagnostics, cluster on standalone, RBAC, audit, query management, edge sync hub/spoke/bundle, and the installation-id header. Skipped unless ARC_TEST_URL is set; extra gates for licensed targets, local storage access and edge-sync hub/spoke. --- tests/integration/test_live_parity.py | 1506 +++++++++++++++++++++++++ 1 file changed, 1506 insertions(+) create mode 100644 tests/integration/test_live_parity.py diff --git a/tests/integration/test_live_parity.py b/tests/integration/test_live_parity.py new file mode 100644 index 0000000..e9f322b --- /dev/null +++ b/tests/integration/test_live_parity.py @@ -0,0 +1,1506 @@ +"""Live parity tests for the 0.2.0 SDK surface against a real Arc server. + +Opt-in. Every test is skipped unless ``ARC_TEST_URL`` is set. They write, +delete, back up and restore data, so point them at a throwaway server only. + +Environment: + +``ARC_TEST_URL`` / ``ARC_TEST_TOKEN`` + The Arc server and an admin token. Required for everything. +``ARC_TEST_LICENSED=1`` + The server has an enterprise license with governance, RBAC, audit, + query management, backup and compaction enabled. Gates those tests. +``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 + files Arc wrote (``arc:tags`` metadata). +``ARC_TEST_HUB_URL`` / ``ARC_TEST_HUB_TOKEN`` + An Arc started with ``edge_sync.enabled`` and ``edge_sync.import.enabled``. +``ARC_TEST_SPOKE_URL`` / ``ARC_TEST_SPOKE_TOKEN`` / ``ARC_TEST_SPOKE_ID`` + An Arc started with ``edge_sync.spoke.enabled`` and + ``edge_sync.spoke.bundle.enabled``, registered on the hub under + ``ARC_TEST_SPOKE_ID`` (default ``spoke-live``). +``ARC_TEST_EDGE_BUNDLE_DIR`` + A directory inside both the spoke's ``edge_sync.spoke.bundle.allowed_dirs`` + and the hub's ``edge_sync.import.allowed_dirs`` (same filesystem). + +Run:: + + ARC_TEST_URL=http://localhost:18081 ARC_TEST_TOKEN= ARC_TEST_LICENSED=1 \\ + ARC_TEST_STORAGE_DIR=/data/arc \\ + ARC_TEST_HUB_URL=http://localhost:18082 ARC_TEST_HUB_TOKEN= \\ + ARC_TEST_SPOKE_URL=http://localhost:18083 ARC_TEST_SPOKE_TOKEN= \\ + ARC_TEST_EDGE_BUNDLE_DIR= \\ + uv run pytest tests/integration/test_live_parity.py -v + +The edge-sync hub needs ``ARC_ENCRYPTION_KEY`` (32 bytes, base64) and +``edge_sync.hub_id``; the spoke needs ``ARC_EDGE_SYNC_SPOKE_SECRET`` (from +``client.edge_sync.register_spoke`` on the hub), ``ARC_EDGE_SYNC_HUB_TOKEN`` +(a hub token with write permission), ``edge_sync.spoke.hub_url``, +``spoke_id`` and ``hub_id``. Keep the spoke's ``sync_interval`` long (e.g. +``1h``) so nothing syncs between the steps of a test. + +Tests that pin down a behavior the SDK merely passes through (e.g. cluster +reads on a standalone node) say so in their docstring. +""" + +from __future__ import annotations + +import contextlib +import glob +import os +import threading +import time +import uuid +from typing import Any, Callable, Iterator, Optional +from urllib.parse import urlparse + +import httpx +import pytest + +from arc_client import ArcClient, AsyncArcClient +from arc_client.exceptions import ( + ArcAuthenticationError, + ArcError, + ArcNotFoundError, + ArcPartialWriteError, + ArcPermissionError, + ArcQueryError, + ArcRateLimitError, + ArcServerError, + ArcValidationError, +) + +ARC_URL = os.environ.get("ARC_TEST_URL") +ARC_TOKEN = os.environ.get("ARC_TEST_TOKEN") +LICENSED = os.environ.get("ARC_TEST_LICENSED") == "1" +STORAGE_DIR = os.environ.get("ARC_TEST_STORAGE_DIR") +HUB_URL = os.environ.get("ARC_TEST_HUB_URL") +HUB_TOKEN = os.environ.get("ARC_TEST_HUB_TOKEN") +SPOKE_URL = os.environ.get("ARC_TEST_SPOKE_URL") +SPOKE_TOKEN = os.environ.get("ARC_TEST_SPOKE_TOKEN") +SPOKE_ID = os.environ.get("ARC_TEST_SPOKE_ID", "spoke-live") +BUNDLE_DIR = os.environ.get("ARC_TEST_EDGE_BUNDLE_DIR") + +pytestmark = pytest.mark.skipif( + not ARC_URL, + reason="Set ARC_TEST_URL (and ARC_TEST_TOKEN) to run live parity tests.", +) + +licensed = pytest.mark.skipif( + not LICENSED, reason="Set ARC_TEST_LICENSED=1 for enterprise-licensed targets." +) +needs_storage = pytest.mark.skipif( + not STORAGE_DIR, reason="Set ARC_TEST_STORAGE_DIR to the server's local storage path." +) +needs_hub = pytest.mark.skipif( + not (HUB_URL and HUB_TOKEN), reason="Set ARC_TEST_HUB_URL/ARC_TEST_HUB_TOKEN." +) +needs_spoke = pytest.mark.skipif( + not (HUB_URL and HUB_TOKEN and SPOKE_URL and SPOKE_TOKEN), + reason="Set ARC_TEST_HUB_URL/_TOKEN and ARC_TEST_SPOKE_URL/_TOKEN.", +) +needs_bundle = pytest.mark.skipif( + not (HUB_URL and HUB_TOKEN and SPOKE_URL and SPOKE_TOKEN and BUNDLE_DIR), + reason="Set the hub/spoke variables and ARC_TEST_EDGE_BUNDLE_DIR.", +) + +INSTALLATION_HEADER = "Arc-Client-Installation-Id" +FLUSH_TIMEOUT = 20.0 + + +# -------------------------------------------------------------------------- +# Helpers +# -------------------------------------------------------------------------- + + +def _target(url: Optional[str]) -> dict[str, Any]: + parsed = urlparse(url or "http://localhost:8000") + ssl = parsed.scheme == "https" + return { + "host": parsed.hostname or "localhost", + "port": parsed.port or (443 if ssl else 80), + "ssl": ssl, + } + + +def make_client(url: Optional[str] = None, token: Optional[str] = None, **kwargs: Any) -> ArcClient: + return ArcClient(**_target(url or ARC_URL), token=token or ARC_TOKEN, **kwargs) + + +def make_async_client( + url: Optional[str] = None, token: Optional[str] = None, **kwargs: Any +) -> AsyncArcClient: + return AsyncArcClient(**_target(url or ARC_URL), token=token or ARC_TOKEN, **kwargs) + + +def uid(prefix: str) -> str: + return f"{prefix}_{uuid.uuid4().hex[:10]}" + + +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() + + +def wait_rows( + client: ArcClient, + sql: str, + database: str, + expected: int, + timeout: float = FLUSH_TIMEOUT, + url: Optional[str] = None, + token: Optional[str] = None, +) -> list[list[Any]]: + """Flush, then poll ``sql`` until it returns ``expected`` rows (or time out).""" + deadline = time.monotonic() + timeout + rows: list[list[Any]] = [] + while True: + server_flush(url, token) + try: + rows = client.query.query(sql, database).data + except ArcQueryError: + rows = [] # table may not exist until the first flush lands + if len(rows) >= expected or time.monotonic() >= deadline: + return rows + time.sleep(0.3) + + +class RequestCounter: + """Count HTTP requests an SDK client sends, via an httpx event hook.""" + + def __init__(self, client: ArcClient) -> None: + self.requests: list[httpx.Request] = [] + http_client = client._get_http()._get_client() + http_client.event_hooks["request"].append(self.requests.append) + + def count(self, path_fragment: str = "") -> int: + return sum(1 for r in self.requests if path_fragment in r.url.path) + + +@contextlib.contextmanager +def timed() -> Iterator[Callable[[], float]]: + start = time.monotonic() + yield lambda: time.monotonic() - start + + +def parquet_files(database: str, measurement: str) -> list[str]: + assert STORAGE_DIR + return glob.glob( + os.path.join(STORAGE_DIR, database, measurement, "**", "*.parquet"), recursive=True + ) + + +# -------------------------------------------------------------------------- +# Fixtures +# -------------------------------------------------------------------------- + + +@pytest.fixture +def admin() -> Iterator[ArcClient]: + with make_client() as client: + yield client + + +@pytest.fixture +async def aadmin() -> Any: + async with make_async_client() as client: + yield client + + +@pytest.fixture +def db(admin: ArcClient) -> Iterator[str]: + """A fresh database, deleted afterwards.""" + name = uid("sdk_lp") + admin.databases.create(name) + yield name + with contextlib.suppress(ArcError): + admin.databases.delete(name, confirm=True) + + +class TokenFactory: + def __init__(self, admin: ArcClient) -> None: + self.admin = admin + self.ids: list[int] = [] + + def create(self, permissions: Optional[list[str]], **kwargs: Any) -> tuple[int, str]: + name = uid("sdk_tok") + created = self.admin.auth.create_token(name, permissions=permissions, **kwargs) + assert created.success and created.token + token_id = next(t.id for t in self.admin.auth.list_tokens() if t.name == name) + self.ids.append(token_id) + return token_id, created.token + + def cleanup(self) -> None: + for token_id in self.ids: + with contextlib.suppress(ArcError): + self.admin.auth.delete_token(token_id) + + +@pytest.fixture +def tokens(admin: ArcClient) -> Iterator[TokenFactory]: + factory = TokenFactory(admin) + yield factory + factory.cleanup() + + +@pytest.fixture +def seeded(admin: ArcClient, db: str) -> tuple[str, str, int]: + """A measurement with three rows (tag ``host``), flushed and queryable.""" + measurement = uid("cpu") + base = (int(time.time()) - 60) * 1_000_000 + admin.write.write_columnar( + measurement, + { + "time": [base, base + 1_000_000, base + 2_000_000], + "host": ["a", "a", "b"], + "usage": [1.0, 2.0, 3.0], + }, + database=db, + tag_columns=["host"], + ) + rows = wait_rows(admin, f"SELECT * FROM {measurement}", db, 3) + assert len(rows) == 3 + return db, measurement, base + + +# -------------------------------------------------------------------------- +# Health and readiness +# -------------------------------------------------------------------------- + + +def test_health_model_fields(admin: ArcClient) -> None: + health = admin.health() + assert health.status == "ok" + assert health.time + assert health.uptime and health.uptime_sec and health.uptime_sec > 0 + assert health.storage is not None and "hot" in health.storage + assert health.storage["hot"].state == "ok" + assert health.storage["hot"].backend + if LICENSED: + assert health.license is not None + assert health.license.status == "active" + assert health.license.tier + assert health.license.days_remaining > 0 + + +def test_readiness_and_write_readiness(admin: ArcClient) -> None: + ready = admin.readiness() + assert ready.ready is True + assert ready.status == "ready" + assert ready.uptime_sec and ready.uptime_sec > 0 + write_ready = admin.write_readiness() + assert write_ready.ready is True + assert write_ready.status == "ready" + assert admin.ready() is True + + +def test_ready_false_when_unreachable() -> None: + with ArcClient(host="127.0.0.1", port=1, max_retries=0, timeout=2) as client, timed() as t: + assert client.ready() is False + assert t() < 5 + + +async def test_async_health_and_readiness(aadmin: AsyncArcClient) -> None: + health = await aadmin.health() + assert health.status == "ok" and health.storage + assert (await aadmin.readiness()).ready is True + assert (await aadmin.write_readiness()).status == "ready" + assert await aadmin.ready() is True + + +# -------------------------------------------------------------------------- +# Query +# -------------------------------------------------------------------------- + + +def test_query_json_profile_and_query_id(admin: ArcClient, seeded: tuple[str, str, int]) -> None: + db, measurement, _ = seeded + result = admin.query.query(f"SELECT host, usage FROM {measurement} ORDER BY time", db) + assert result.row_count == 3 + assert result.data == [["a", 1.0], ["a", 2.0], ["b", 3.0]] + assert result.query_id, "X-Arc-Query-ID not surfaced on JSON query" + assert result.profile is None + + profiled = admin.query.query(f"SELECT count(*) FROM {measurement}", db, profile=True) + assert profiled.profile is not None, "profile=True returned no profile" + assert profiled.profile.total_ms is not None and profiled.profile.total_ms >= 0 + + +def test_query_msgpack_profile_and_query_id(admin: ArcClient, seeded: tuple[str, str, int]) -> None: + db, measurement, _ = seeded + result = admin.query.query_msgpack( + f"SELECT host, usage FROM {measurement} ORDER BY time", db, profile=True + ) + assert result.columns == ["host", "usage"] + assert result.data == [["a", 1.0], ["a", 2.0], ["b", 3.0]] + assert result.query_id, "X-Arc-Query-ID not surfaced on msgpack query" + assert result.profile is not None, "profile=True returned no profile on msgpack" + + +def test_query_arrow_metadata_dictionary_compression( + admin: ArcClient, seeded: tuple[str, str, int] +) -> None: + import pyarrow as pa + + db, measurement, _ = seeded + sql = f"SELECT host, usage FROM {measurement} ORDER BY time" + plain = admin.query.query_arrow(sql, db) + assert plain.num_rows == 3 + meta = plain.schema.metadata or {} + assert meta.get(b"arc.query_id"), f"arc.query_id missing from Arrow metadata: {meta}" + assert not pa.types.is_dictionary(plain.schema.field("host").type) + + # Arc dictionary-encodes only from 256 rows and <= rows/4 uniques + # (internal/api/query_arrow_dict.go dictMinRows). + small = admin.query.query_arrow(sql, db, dictionary=True) + assert small.column("host").to_pylist() == ["a", "a", "b"] + many = "SELECT 'h' || (i % 3)::VARCHAR AS host, i FROM range(1000) t(i) ORDER BY i" + dictionary = admin.query.query_arrow(many, db, dictionary=True) + assert pa.types.is_dictionary(dictionary.schema.field("host").type), dictionary.schema + assert dictionary.column("host").to_pylist()[:4] == ["h0", "h1", "h2", "h0"] + assert not pa.types.is_dictionary(admin.query.query_arrow(many, db).schema.field("host").type) + + for codec in ("zstd", "lz4"): + compressed = admin.query.query_arrow(sql, db, compression=codec) + assert compressed.column("usage").to_pylist() == [1.0, 2.0, 3.0] + assert compressed.column("host").to_pylist() == ["a", "a", "b"] + + with pytest.raises(ArcValidationError): + admin.query.query_arrow(sql, db, compression="gzip") + + +def test_query_pandas_and_polars(admin: ArcClient, seeded: tuple[str, str, int]) -> None: + db, measurement, _ = seeded + sql = f"SELECT host, usage FROM {measurement} ORDER BY time" + df = admin.query.query_pandas(sql, db, dictionary=True, compression="zstd") + assert list(df["host"].astype(str)) == ["a", "a", "b"] + assert list(df["usage"]) == [1.0, 2.0, 3.0] + pl_df = admin.query.query_polars(sql, db, compression="lz4") + assert pl_df["usage"].to_list() == [1.0, 2.0, 3.0] + + +def test_query_measurement(admin: ArcClient, seeded: tuple[str, str, int]) -> None: + db, measurement, _ = seeded + result = admin.query.query_measurement( + measurement, db, limit=2, order_by="time ASC", where="host = 'a'" + ) + usage = result.columns.index("usage") + assert [row[usage] for row in result.data] == [1.0, 2.0] + offset = admin.query.query_measurement(measurement, db, limit=10, offset=2, order_by="time ASC") + assert offset.row_count == 1 + default = admin.query.query_measurement(measurement, db) + assert default.row_count == 3 + + +def test_estimate(admin: ArcClient, seeded: tuple[str, str, int]) -> None: + db, measurement, _ = seeded + estimate = admin.query.estimate(f"SELECT * FROM {measurement}", db) + assert estimate.success + assert estimate.estimated_rows is not None + + +@licensed +def test_query_id_lookup_and_kill_long_query(admin: ArcClient) -> None: + """A finished query is in history by its query_id; a running one can be killed.""" + finished = admin.query.query("SELECT 42 AS answer") + assert finished.query_id + info = admin.queries.get(finished.query_id) + assert finished.query_id in str(info) + + sql = "SELECT sum(i % 7) FROM range(400000000000) t(i)" + outcome: dict[str, Any] = {} + + def run() -> None: + with make_client(timeout=120, max_retries=0) as long_client: + try: + outcome["result"] = long_client.query.query(sql) + except BaseException as e: # noqa: BLE001 + outcome["error"] = e + + thread = threading.Thread(target=run, daemon=True) + thread.start() + query_id = None + deadline = time.monotonic() + 15 + while query_id is None and time.monotonic() < deadline: + for q in admin.queries.active().get("queries") or []: + if "range(400000000000)" in str(q.get("sql", "")): + query_id = q.get("id") or q.get("query_id") + time.sleep(0.2) + assert query_id, "long query never showed up in queries.active()" + killed = admin.queries.kill(query_id) + assert killed + thread.join(30) + assert not thread.is_alive(), "killed query did not return" + assert "error" in outcome, f"killed query returned a result: {outcome}" + assert isinstance(outcome["error"], ArcError) + + +async def test_async_query_surfaces(aadmin: AsyncArcClient, seeded: tuple[str, str, int]) -> None: + db, measurement, _ = seeded + result = await aadmin.query.query( + f"SELECT usage FROM {measurement} ORDER BY time", db, profile=True + ) + assert result.data == [[1.0], [2.0], [3.0]] + assert result.query_id and result.profile is not None + table = await aadmin.query.query_arrow(f"SELECT host FROM {measurement}", db, dictionary=True) + assert (table.schema.metadata or {}).get(b"arc.query_id") + by_name = await aadmin.query.query_measurement(measurement, db, limit=1) + assert by_name.row_count == 1 + + +# -------------------------------------------------------------------------- +# Governance: a 429 must be raised at once, never retried +# -------------------------------------------------------------------------- + + +def _rate_limited_client( + admin: ArcClient, tokens: TokenFactory +) -> tuple[ArcClient, RequestCounter, int]: + token_id, token = tokens.create(["read"]) + admin.governance.create_policy(token_id, rate_limit_per_minute=1) + client = make_client(token=token, max_retries=3, retry_backoff=2.0) + return client, RequestCounter(client), token_id + + +@licensed +@pytest.mark.parametrize("surface", ["query", "query_arrow", "query_msgpack", "estimate"]) +def test_governance_rate_limit_raised_not_retried( + admin: ArcClient, tokens: TokenFactory, seeded: tuple[str, str, int], surface: str +) -> None: + db, measurement, _ = seeded + client, counter, token_id = _rate_limited_client(admin, tokens) + sql = f"SELECT count(*) FROM {measurement}" + try: + call = getattr(client.query, surface) + call(sql, db) + with timed() as t, pytest.raises(ArcRateLimitError): + call(sql, db) + assert counter.count("/api/v1/query") == 2, [r.url for r in counter.requests] + assert t() < 1.5 + finally: + client.close() + with contextlib.suppress(ArcError): + admin.governance.delete_policy(token_id) + + +@licensed +def test_governance_rate_limit_on_query_measurement_not_retried( + admin: ArcClient, tokens: TokenFactory, seeded: tuple[str, str, int] +) -> None: + """GET /api/v1/query/:measurement is governed too (Arc query.go checkQueryGovernance). + + The SDK retries 429 on every GET, so a governance 429 here is retried, + sleeping Retry-After (up to 60 s) between attempts. + """ + db, measurement, _ = seeded + client, counter, token_id = _rate_limited_client(admin, tokens) + sleeps: list[float] = [] + import arc_client.http.sync_http as sync_http + + real_sleep = sync_http.time.sleep + try: + client.query.query_measurement(measurement, db, limit=1) + sync_http.time.sleep = sleeps.append # type: ignore[assignment] + with pytest.raises(ArcRateLimitError): + client.query.query_measurement(measurement, db, limit=1) + assert counter.count(f"/api/v1/query/{measurement}") == 2, ( + f"governance 429 retried: {counter.count('/api/v1/query/')} requests, sleeps={sleeps}" + ) + finally: + sync_http.time.sleep = real_sleep # type: ignore[assignment] + client.close() + with contextlib.suppress(ArcError): + admin.governance.delete_policy(token_id) + + +@licensed +def test_governance_policy_crud_and_usage(admin: ArcClient, tokens: TokenFactory) -> None: + token_id, _ = tokens.create(["read"]) + created = admin.governance.create_policy( + token_id, rate_limit_per_minute=10, max_rows_per_query=5 + ) + assert created + policy = admin.governance.get_policy(token_id) + assert "10" in str(policy) and str(token_id) in str(policy) + admin.governance.update_policy(token_id, rate_limit_per_minute=20) + assert "20" in str(admin.governance.get_policy(token_id)) + listed = admin.governance.list_policies() + assert any(p.get("token_id") == token_id for p in listed.get("policies", [])) + usage = admin.governance.usage(token_id) + assert str(token_id) in str(usage) + admin.governance.delete_policy(token_id) + with pytest.raises(ArcNotFoundError): + admin.governance.get_policy(token_id) + + +# -------------------------------------------------------------------------- +# Write +# -------------------------------------------------------------------------- + + +def test_line_protocol_precision(admin: ArcClient, db: str) -> None: + measurement = uid("lp_prec") + seconds = int(time.time()) - 120 + admin.write.write_line_protocol( + f"{measurement},host=a v=1i {seconds}", database=db, precision="s" + ) + admin.write.write_point( + measurement, + {"v": 2}, + tags={"host": "b"}, + timestamp=(seconds + 1) * 1000, + database=db, + precision="ms", + ) + rows = wait_rows(admin, f"SELECT host, epoch_us(time) FROM {measurement} ORDER BY time", db, 2) + assert rows == [["a", seconds * 1_000_000], ["b", (seconds + 1) * 1_000_000]] + + with pytest.raises(ArcValidationError): + admin.write.write_line_protocol(f"{measurement} v=1i {seconds}", database=db, precision="h") + + +def _arc_tags(database: str, measurement: str) -> set[str]: + import pyarrow.parquet as pq + + files = parquet_files(database, measurement) + assert files, f"no parquet files for {database}/{measurement} under {STORAGE_DIR}" + tags: set[str] = set() + for path in files: + meta = pq.read_schema(path).metadata or {} + raw = meta.get(b"arc:tags") + if raw: + tags.update(t for t in raw.decode().split(",") if t) + return tags + + +@needs_storage +def test_columnar_tag_columns_written_as_tags(admin: ArcClient, db: str) -> None: + with_tags = uid("col_tags") + without = uid("col_notags") + base = (int(time.time()) - 60) * 1_000_000 + cols = {"time": [base, base + 1], "host": ["a", "b"], "region": ["x", "y"], "v": [1.0, 2.0]} + admin.write.write_columnar(with_tags, dict(cols), database=db, tag_columns=["host", "region"]) + admin.write.write_columnar(without, dict(cols), database=db) + wait_rows(admin, f"SELECT * FROM {with_tags}", db, 2) + wait_rows(admin, f"SELECT * FROM {without}", db, 2) + assert _arc_tags(db, with_tags) == {"host", "region"} + assert _arc_tags(db, without) == set() + + +@needs_storage +def test_dataframe_tag_columns_written_as_tags(admin: ArcClient, db: str) -> None: + import pandas as pd + + measurement = uid("df_tags") + now = pd.Timestamp.now(tz="UTC").floor("s") - pd.Timedelta(minutes=1) + df = pd.DataFrame( + {"time": [now, now + pd.Timedelta(seconds=1)], "host": ["a", "b"], "v": [1.5, 2.5]} + ) + admin.write.write_dataframe(df, measurement, database=db, tag_columns=["host"]) + rows = wait_rows(admin, f"SELECT host, v FROM {measurement} ORDER BY time", db, 2) + assert rows == [["a", 1.5], ["b", 2.5]] + assert _arc_tags(db, measurement) == {"host"} + + +def test_write_records(admin: ArcClient, db: str) -> None: + measurement = uid("rec") + base = (int(time.time()) - 60) * 1_000_000 + admin.write.write_records( + [ + { + "measurement": measurement, + "timestamp": base, + "tags": {"host": "a"}, + "fields": {"v": 1.0}, + }, + { + "measurement": measurement, + "timestamp": base + 1, + "tags": {"host": "b"}, + "fields": {"v": 2.0}, + }, + ], + database=db, + ) + rows = wait_rows(admin, f"SELECT host, v FROM {measurement} ORDER BY time", db, 2) + assert rows == [["a", 1.0], ["b", 2.0]] + + +def test_buffered_writer(admin: ArcClient, db: str) -> None: + measurement = uid("buf") + base = (int(time.time()) - 60) * 1_000_000 + client = make_client(database=db) + with client, client.write.buffered(batch_size=1000, flush_interval=3600) as buffer: + for i in range(5): + buffer.write( + { + "measurement": measurement, + "timestamp": base + i, + "tags": {"host": "a"}, + "fields": {"v": float(i)}, + } + ) + buffer.write_columnar( + measurement, {"time": [base + 10], "host": ["b"], "v": [10.0]}, tag_columns=["host"] + ) + assert buffer.pending_count == 6 + rows = wait_rows(admin, f"SELECT host, v FROM {measurement} ORDER BY time", db, 6) + assert [r[1] for r in rows] == [0.0, 1.0, 2.0, 3.0, 4.0, 10.0] + + +def test_partial_write_line_protocol(admin: ArcClient, db: str) -> None: + """A type conflict in one LP body: 400 "partial write", the rest stored once.""" + measurement = uid("pw") + s = int(time.time()) - 60 + body = [ + f"{measurement},host=a v=1i {s}", + f'{measurement},host=b v="oops" {s + 1}', + f"{measurement},host=c v=3i {s + 2}", + ] + with make_client(max_retries=3, retry_backoff=2.0) as client: + counter = RequestCounter(client) + with timed() as t, pytest.raises(ArcPartialWriteError) as excinfo: + client.write.write_line_protocol(body, database=db, precision="s") + assert excinfo.value.dropped == 1 + assert excinfo.value.others_written is True + assert counter.count("/api/v1/write/line-protocol") == 1 + assert t() < 1.5 + rows = wait_rows(admin, f"SELECT host, v FROM {measurement} ORDER BY time", db, 2) + time.sleep(1) + rows = wait_rows(admin, f"SELECT host, v FROM {measurement} ORDER BY time", db, 2) + assert rows == [["a", 1], ["c", 3]] + + +def test_partial_write_nothing_written(admin: ArcClient, db: str) -> None: + measurement = uid("pw_none") + s = int(time.time()) - 60 + with pytest.raises(ArcPartialWriteError) as excinfo: + admin.write.write_line_protocol( + f"{measurement},host=a time=1i,v=2i {s}", database=db, precision="s" + ) + assert excinfo.value.dropped == 1 + assert excinfo.value.others_written is False + + +def test_partial_write_msgpack_records(admin: ArcClient, db: str) -> None: + """MessagePack row format follows line protocol's drop rules (Arc msgpack.go).""" + measurement = uid("pw_rec") + base = (int(time.time()) - 60) * 1_000_000 + with pytest.raises(ArcPartialWriteError) as excinfo: + admin.write.write_records( + [ + {"measurement": measurement, "timestamp": base, "fields": {"v": 1}}, + {"measurement": measurement, "timestamp": base + 1, "fields": {"v": "oops"}}, + {"measurement": measurement, "timestamp": base + 2, "fields": {"v": 3}}, + ], + database=db, + ) + assert excinfo.value.dropped == 1 and excinfo.value.others_written is True + rows = wait_rows(admin, f"SELECT v FROM {measurement} ORDER BY time", db, 2) + assert rows == [[1], [3]] + + +def test_partial_write_buffered_reported_once(admin: ArcClient, db: str) -> None: + """A buffered batch Arc drops (partial write) is discarded and reported once. + + The buffered writer sends columnar MessagePack, where Arc drops a whole + record whose names it cannot read back (a field named "time" in another + case, internal/api/msgpack.go FilterRecordNames). The other measurement + in the same flush is still written, exactly once. + """ + dropped = uid("pw_buf") + good = uid("pw_buf_ok") + base = (int(time.time()) - 60) * 1_000_000 + with make_client(database=db, max_retries=3, retry_backoff=2.0) as client: + counter = RequestCounter(client) + buffer = client.write.buffered(batch_size=1000, flush_interval=3600) + buffer.write({"measurement": dropped, "timestamp": base, "fields": {"v": 1.0, "TIME": 5.0}}) + buffer.write( + {"measurement": dropped, "timestamp": base + 1, "fields": {"v": 2.0, "TIME": 6.0}} + ) + buffer.write({"measurement": good, "timestamp": base, "fields": {"v": 1.0}}) + buffer.write({"measurement": good, "timestamp": base + 1, "fields": {"v": 2.0}}) + with timed() as t, pytest.raises(ArcPartialWriteError) as excinfo: + buffer.flush() + assert t() < 1.5 + assert excinfo.value.dropped == 2 + assert excinfo.value.others_written is False + assert buffer.pending_count == 0, "partial-write batch should be discarded" + buffer.flush() # reported once: a second flush does not raise again + buffer.write({"measurement": good, "timestamp": base + 2, "fields": {"v": 3.0}}) + buffer.close() + writes = counter.count("/api/v1/write") + assert writes == 3, f"expected 2 flush writes + 1 close write, got {writes}" + assert wait_rows(admin, f"SELECT v FROM {good} ORDER BY time", db, 3) == [[1.0], [2.0], [3.0]] + time.sleep(1) + assert wait_rows(admin, f"SELECT v FROM {good} ORDER BY time", db, 3) == [[1.0], [2.0], [3.0]] + try: + stored = admin.query.query(f"SELECT * FROM {dropped}", db).row_count + except ArcQueryError: + stored = 0 + assert stored == 0, "a dropped batch was stored" + + +def test_columnar_type_conflict_status(admin: ArcClient, db: str) -> None: + """A mixed-type column in one columnar batch is a client error. + + Line protocol and MessagePack rows answer it with a 400 partial write; + columnar MessagePack rejects the whole batch. The status should be a 4xx: + a 5xx tells the caller (and the buffered writer) that the server failed. + """ + measurement = uid("col_conflict") + base = (int(time.time()) - 60) * 1_000_000 + with pytest.raises(ArcError) as excinfo: + admin.write.write_columnar( + measurement, {"time": [base, base + 1], "v": [1, "oops"]}, database=db + ) + status = getattr(excinfo.value, "status_code", None) or getattr( + excinfo.value.__cause__, "status_code", None + ) + assert status is not None and 400 <= status < 500, f"status {status}: {excinfo.value}" + + +async def test_async_write_and_partial(aadmin: AsyncArcClient, admin: ArcClient, db: str) -> None: + measurement = uid("apw") + s = int(time.time()) - 60 + with pytest.raises(ArcPartialWriteError) as excinfo: + await aadmin.write.write_line_protocol( + [f"{measurement} v=1i {s}", f'{measurement} v="x" {s + 1}'], database=db, precision="s" + ) + assert excinfo.value.dropped == 1 and excinfo.value.others_written + await aadmin.write.write_columnar( + measurement + "_c", + {"time": [s * 1_000_000], "host": ["z"], "v": [9.0]}, + database=db, + tag_columns=["host"], + ) + assert len(wait_rows(admin, f"SELECT * FROM {measurement}_c", db, 1)) == 1 + + +# -------------------------------------------------------------------------- +# Databases and delete +# -------------------------------------------------------------------------- + + +def test_measurement_schema_get_and_rebuild(admin: ArcClient, seeded: tuple[str, str, int]) -> None: + db, measurement, _ = seeded + schema = admin.databases.get_measurement_schema(db, measurement) + assert schema.database == db and schema.measurement == measurement + names = {f.name: f.type for f in schema.fields} + assert {"time", "host", "usage"} <= set(names), names + rebuilt = admin.databases.rebuild_measurement_schema(db, measurement) + assert rebuilt.status == "queued" + assert rebuilt.database == db and rebuilt.measurement == measurement + with pytest.raises(ArcNotFoundError): + admin.databases.get_measurement_schema(db, uid("missing")) + + +def test_delete_allow_unreachable_on_standalone( + admin: ArcClient, seeded: tuple[str, str, int] +) -> None: + db, measurement, _ = seeded + preview = admin.delete.delete( + db, measurement, "host = 'a'", dry_run=True, allow_unreachable=True + ) + assert preview.dry_run is True + assert preview.deleted_count == 2 + done = admin.delete.delete( + db, measurement, "host = 'a'", dry_run=False, confirm=True, allow_unreachable=True + ) + assert done.deleted_count == 2 + assert not done.unreachable_files and not done.unflushed_nodes + rows = admin.query.query(f"SELECT host FROM {measurement}", db).data + assert rows == [["b"]] + + +def test_databases_delete_allow_unreachable(admin: ArcClient) -> None: + name = uid("sdk_dbdel") + admin.databases.create(name) + admin.write.write_columnar("m", {"time": [int(time.time() * 1e6)], "v": [1.0]}, database=name) + server_flush() + with pytest.raises(ArcValidationError): + admin.databases.delete(name) + admin.databases.delete(name, confirm=True, allow_unreachable=True) + with pytest.raises(ArcNotFoundError): + admin.databases.get(name) + + +async def test_async_databases(aadmin: AsyncArcClient, seeded: tuple[str, str, int]) -> None: + db, measurement, _ = seeded + schema = await aadmin.databases.get_measurement_schema(db, measurement) + assert any(f.name == "usage" for f in schema.fields) + assert (await aadmin.databases.get(db)).name == db + + +# -------------------------------------------------------------------------- +# Auth +# -------------------------------------------------------------------------- + + +def test_create_token_with_empty_permissions(admin: ArcClient, tokens: TokenFactory) -> None: + token_id, token = tokens.create([]) + info = admin.auth.get_token(token_id) + assert info.permissions == [], f"empty permissions not kept: {info.permissions}" + with make_client(token=token) as client: + verified = client.auth.verify() + assert verified.valid + assert not verified.permissions, verified + with pytest.raises(ArcAuthenticationError): + client.query.query("SELECT 1") + + +def test_update_token_to_empty_permissions(admin: ArcClient, tokens: TokenFactory) -> None: + token_id, token = tokens.create(["read"]) + with make_client(token=token) as client: + assert client.query.query("SELECT 1").row_count == 1 + admin.auth.update_token(token_id, permissions=[]) + assert admin.auth.get_token(token_id).permissions == [] + admin.auth.invalidate_cache() + with make_client(token=token) as client, pytest.raises(ArcAuthenticationError): + client.query.query("SELECT 1") + + +def test_auth_cache_stats_and_invalidate(admin: ArcClient) -> None: + stats = admin.auth.cache_stats() + assert "cache_size" in stats and "hit_rate_percent" in stats, stats + invalidated = admin.auth.invalidate_cache() + assert invalidated.get("success") is not False, invalidated + assert admin.auth.cache_stats()["cache_size"] <= stats["max_cache_size"] + + +def test_auth_cache_requires_admin(tokens: TokenFactory) -> None: + _, token = tokens.create(["read"]) + with make_client(token=token) as client, pytest.raises(ArcPermissionError): + client.auth.cache_stats() + + +async def test_async_auth(aadmin: AsyncArcClient, tokens: TokenFactory) -> None: + token_id, _ = tokens.create([]) + info = await aadmin.auth.get_token(token_id) + assert info.permissions == [] + await aadmin.auth.update_token(token_id, permissions=["read"]) + assert (await aadmin.auth.get_token(token_id)).permissions == ["read"] + assert "cache_size" in await aadmin.auth.cache_stats() + + +# -------------------------------------------------------------------------- +# Compaction +# -------------------------------------------------------------------------- + + +@licensed +def test_compaction_reads(admin: ArcClient) -> None: + status = admin.compaction.status() + assert "manager" in status + cycles = admin.compaction.cycles(limit=5) + assert "cycles" in cycles + too_wide = admin.compaction.too_wide() + assert "partitions" in too_wide + cleared = admin.compaction.clear_too_wide() + assert isinstance(cleared, dict) + assert isinstance(admin.compaction.history(limit=5), (dict, list)) + with pytest.raises(ArcNotFoundError): + admin.compaction.cycle(987654321) + + +@licensed +def test_compaction_trigger_measurement(admin: ArcClient, db: str) -> None: + """Write several old files into one hour partition, then compact that measurement.""" + measurement = uid("cmp") + hour = (int(time.time()) // 3600 - 30) * 3600 + for i in range(4): + admin.write.write_columnar( + measurement, + {"time": [(hour + i * 60) * 1_000_000], "host": [f"h{i}"], "v": [float(i)]}, + database=db, + tag_columns=["host"], + ) + server_flush() + assert len(wait_rows(admin, f"SELECT * FROM {measurement}", db, 4)) == 4 + + with pytest.raises(ArcValidationError): + admin.compaction.trigger(measurement=measurement) + before = { + c.get("cycle_id") or c.get("id") for c in admin.compaction.cycles().get("cycles") or [] + } + triggered = admin.compaction.trigger(tier="hourly", database=db, measurement=measurement) + assert triggered + cycle = None + deadline = time.monotonic() + 60 + while cycle is None and time.monotonic() < deadline: + for c in admin.compaction.cycles().get("cycles") or []: + cid = c.get("cycle_id") or c.get("id") + if cid not in before and c.get("state", c.get("status")) not in ("running", "pending"): + cycle = c + time.sleep(1) + assert cycle, f"no finished compaction cycle after trigger: {admin.compaction.cycles()}" + cycle_id = cycle.get("cycle_id") or cycle.get("id") + detail = admin.compaction.cycle(cycle_id) + assert str(cycle_id) in str(detail) + history = admin.compaction.history(cycle_id=cycle_id) + assert isinstance(history, (dict, list)) + rows = admin.query.query(f"SELECT v FROM {measurement} ORDER BY time", db).data + assert rows == [[0.0], [1.0], [2.0], [3.0]] + + +async def test_async_compaction(aadmin: AsyncArcClient) -> None: + if not LICENSED: + pytest.skip("Set ARC_TEST_LICENSED=1") + assert "cycles" in await aadmin.compaction.cycles(limit=1) + assert "partitions" in await aadmin.compaction.too_wide() + + +# -------------------------------------------------------------------------- +# Backup +# -------------------------------------------------------------------------- + + +def _wait_backup_done(admin: ArcClient, timeout: float = 60) -> dict[str, Any]: + deadline = time.monotonic() + timeout + status: dict[str, Any] = {} + while time.monotonic() < deadline: + status = admin.backup.status() + if status.get("status") != "running": + return status + time.sleep(0.5) + raise AssertionError(f"backup never went idle: {status}") + + +@licensed +def test_backup_scoped_restore_merge_and_replace( + admin: ArcClient, seeded: tuple[str, str, int] +) -> None: + db, measurement, _ = seeded + _wait_backup_done(admin) + started = admin.backup.create(databases=[db]) + assert started.get("status") == "running" + assert started.get("databases") == [db] + _wait_backup_done(admin) + + listing = admin.backup.list(include_foreign=True) + mine = [b for b in listing.get("backups") or [] if db in str(b)] + assert mine, f"scoped backup not listed: {listing}" + backup_id = mine[0].get("backup_id") or mine[0].get("id") + try: + manifest = admin.backup.get(backup_id) + assert db in str(manifest) + + # Delete the data, then merge-restore it back. + admin.delete.delete(db, measurement, "1=1", dry_run=False, confirm=True) + assert admin.query.query(f"SELECT * FROM {measurement}", db).row_count == 0 + merged = admin.backup.restore(backup_id, confirm=True, mode="merge") + assert merged.get("mode") == "merge", merged + _wait_backup_done(admin) + rows = wait_rows(admin, f"SELECT usage FROM {measurement} ORDER BY time", db, 3) + assert rows == [[1.0], [2.0], [3.0]] + + # replace is cluster-only: a standalone server answers 400. + with pytest.raises(ArcServerError) as excinfo: + admin.backup.restore(backup_id, confirm=True, mode="replace") + assert excinfo.value.status_code == 400 + with pytest.raises(ArcValidationError): + admin.backup.restore(backup_id, confirm=True, mode="overwrite") + finally: + _wait_backup_done(admin) + admin.backup.delete(backup_id) + with pytest.raises(ArcNotFoundError): + admin.backup.get(backup_id) + + +# -------------------------------------------------------------------------- +# Continuous queries and retention +# -------------------------------------------------------------------------- + + +def test_continuous_query_lifecycle(admin: ArcClient, seeded: tuple[str, str, int]) -> None: + db, measurement, base = seeded + dest = measurement + "_1m" + cq = admin.continuous_queries.create( + name=uid("cq"), + database=db, + source_measurement=measurement, + destination_measurement=dest, + query=( + f"SELECT time_bucket(INTERVAL '1 minute', time) AS time, host, avg(usage) AS usage " + f"FROM {db}.{measurement} WHERE time >= {{start_time}} AND time < {{end_time}} " + f"GROUP BY 1, host" + ), + interval="1m", + ) + try: + start = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(base / 1e6 - 3600)) + end = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(time.time())) + executed = admin.continuous_queries.execute(cq.id, start_time=start, end_time=end) + assert executed.error is None, executed + # watermark_status is documented as cluster-only (None standalone). + assert executed.watermark_status is None, executed + updated = admin.continuous_queries.update(cq.id, interval="5m") + assert updated.interval == "5m" + replaced = admin.continuous_queries.replace( + cq.id, + name=cq.name, + database=db, + source_measurement=measurement, + destination_measurement=dest, + query=cq.query, + interval="10m", + description="replaced", + tag_columns=None, + retention_days=None, + delete_source_after_days=None, + is_active=True, + ) + assert replaced.interval == "10m" and replaced.description == "replaced" + assert admin.continuous_queries.get_executions(cq.id) + finally: + admin.continuous_queries.delete(cq.id) + + +def test_retention_lifecycle(admin: ArcClient, db: str) -> None: + policy = admin.retention.create(name=uid("ret"), database=db, retention_days=30) + try: + assert admin.retention.get(policy.id).retention_days == 30 + updated = admin.retention.update(policy.id, retention_days=60) + assert updated.retention_days == 60 + assert any(p.id == policy.id for p in admin.retention.list()) + dry = admin.retention.execute(policy.id, dry_run=True) + assert dry is not None + finally: + admin.retention.delete(policy.id) + + +async def test_async_cq_and_retention(aadmin: AsyncArcClient, db: str) -> None: + assert isinstance(await aadmin.continuous_queries.list(database=db), list) + assert isinstance(await aadmin.retention.list(), list) + + +# -------------------------------------------------------------------------- +# Reconciliation +# -------------------------------------------------------------------------- + + +def test_reconciliation_on_standalone(admin: ArcClient) -> None: + status = admin.reconciliation.status() + assert status["enabled"] is False and status.get("reason") + with pytest.raises(ArcServerError) as excinfo: + admin.reconciliation.trigger() + assert excinfo.value.status_code == 503 + with pytest.raises(ArcValidationError): + admin.reconciliation.trigger(dry_run=False) + + +def test_reconciliation_get_run_not_retried_on_standalone(admin: ArcClient) -> None: + """GET .../runs/:id answers 503 "not running" on standalone; the SDK should not stall on it.""" + with make_client(max_retries=3, retry_backoff=2.0) as client: + counter = RequestCounter(client) + with timed() as t, pytest.raises(ArcServerError) as excinfo: + client.reconciliation.get_run("does-not-exist") + assert excinfo.value.status_code == 503 + assert counter.count("/reconciliation/runs/") == 1, ( + f"503 'not running' retried {counter.count('/reconciliation/runs/')} times in {t():.1f}s" + ) + + +# -------------------------------------------------------------------------- +# Diagnostics +# -------------------------------------------------------------------------- + + +def test_diagnostics_reads(admin: ArcClient) -> None: + assert isinstance(admin.diagnostics.metrics(), dict) + assert isinstance(admin.diagnostics.memory_metrics(), dict) + assert isinstance(admin.diagnostics.query_pool_metrics(), dict) + assert isinstance(admin.diagnostics.endpoint_metrics(), dict) + assert isinstance(admin.diagnostics.timeseries("system", duration_minutes=5), dict) + prom = admin.diagnostics.prometheus() + assert isinstance(prom, str) and "# TYPE" in prom + logs = admin.diagnostics.logs(limit=5, level="info", since_minutes=60) + assert isinstance(logs, dict) + assert "heap_alloc" in str(admin.diagnostics.memstats()).lower() or admin.diagnostics.memstats() + assert isinstance(admin.diagnostics.duckdb_memory(), dict) + with pytest.raises(ArcValidationError): + admin.diagnostics.timeseries("bogus") + + +def test_free_os_memory_and_throttle(admin: ArcClient) -> None: + with pytest.raises(ArcValidationError): + admin.diagnostics.free_os_memory() + try: + first = admin.diagnostics.free_os_memory(confirm=True) + except ArcRateLimitError as e: + wait = (e.response_body or {}).get("retry_after_seconds", 30) + time.sleep(float(wait) + 1) + first = admin.diagnostics.free_os_memory(confirm=True) + assert "heap_released_after" in first + with make_client(max_retries=3, retry_backoff=2.0) as client: + counter = RequestCounter(client) + with timed() as t, pytest.raises(ArcRateLimitError) as excinfo: + client.diagnostics.free_os_memory(confirm=True) + assert counter.count("free-os-memory") == 1 + assert t() < 1.5 + assert (excinfo.value.response_body or {}).get("retry_after_seconds") + + +def test_diagnostics_admin_routes_need_admin(tokens: TokenFactory) -> None: + _, token = tokens.create(["read"]) + with make_client(token=token) as client: + assert isinstance(client.diagnostics.metrics(), dict) + with pytest.raises(ArcPermissionError): + client.diagnostics.memstats() + + +async def test_async_diagnostics(aadmin: AsyncArcClient) -> None: + assert isinstance(await aadmin.diagnostics.metrics(), dict) + assert "# TYPE" in await aadmin.diagnostics.prometheus() + assert isinstance(await aadmin.diagnostics.memstats(), dict) + + +# -------------------------------------------------------------------------- +# Cluster methods on a standalone node +# -------------------------------------------------------------------------- + +CLUSTER_READS = ["status", "nodes", "local", "health", "files", "policies"] + + +@pytest.mark.parametrize("method", CLUSTER_READS) +def test_cluster_reads_on_standalone_report_disabled(admin: ArcClient, method: str) -> None: + """Pins down pass-through behavior: these answer 200 ``{"enabled": false, "success": false}``. + + The SDK returns that dict as-is; nothing raises, so ``nodes()`` looks like + a successful call. Callers must check ``enabled``. + """ + result = getattr(admin.cluster, method)() + assert result.get("enabled") is False and result.get("mode") == "standalone", result + + +def test_cluster_node_on_standalone(admin: ArcClient) -> None: + result = admin.cluster.node("no-such-node") + assert result.get("enabled") is False, result + + +def test_cluster_stream_claims_on_standalone(admin: ArcClient) -> None: + with pytest.raises(ArcServerError) as excinfo: + admin.cluster.stream_claims() + assert excinfo.value.status_code == 409 + + +CLUSTER_MUTATIONS: list[tuple[str, Callable[[Any], Any]]] = [ + ("remove_node", lambda c: c.remove_node("no-such-node")), + ("demote_writer", lambda c: c.demote_writer("no-such-node", confirm=True)), + ("assign_compactor", lambda c: c.assign_compactor("no-such-node")), + ("converge_voters", lambda c: c.converge_voters(confirm=True)), + ("release_stream_claim", lambda c: c.release_stream_claim("00000000000000ff", epoch=1)), + ("abort_compaction_job", lambda c: c.abort_compaction_job("no-such-job", confirm=True)), + ("adopt_policies", lambda c: c.adopt_policies("cq", "no-such-node", "0" * 64, confirm=True)), + ("delete_file", lambda c: c.delete_file("db/m/2026/01/01/00/x.parquet", confirm=True)), +] + + +@pytest.mark.parametrize("name,call", CLUSTER_MUTATIONS, ids=[n for n, _ in CLUSTER_MUTATIONS]) +def test_cluster_mutations_on_standalone_fail_clearly( + admin: ArcClient, name: str, call: Callable[[Any], Any] +) -> None: + """A cluster mutation on a standalone node must not look like it succeeded.""" + try: + result = call(admin.cluster) + except ArcError as e: + print(f"cluster.{name}: {type(e).__name__} status={getattr(e, 'status_code', None)} {e}") + return + print(f"cluster.{name}: returned {result}") + assert result.get("success") is not True and result.get("enabled") is False, result + + +# -------------------------------------------------------------------------- +# Enterprise: RBAC, audit, query management +# -------------------------------------------------------------------------- + + +@licensed +def test_rbac_full_lifecycle( + admin: ArcClient, tokens: TokenFactory, seeded: tuple[str, str, int] +) -> None: + db, measurement, _ = seeded + org = admin.rbac.create_organization(uid("org"), description="sdk live") + org_id = (org.get("organization") or org)["id"] + try: + assert admin.rbac.get_organization(org_id) + admin.rbac.update_organization(org_id, description="updated") + assert any(o["id"] == org_id for o in admin.rbac.list_organizations()["organizations"]) + + team = admin.rbac.create_team(org_id, uid("team")) + team_id = (team.get("team") or team)["id"] + assert admin.rbac.get_team(team_id) # /api/v1/rbac/teams/:id + admin.rbac.update_team(team_id, description="t") + assert any(t["id"] == team_id for t in admin.rbac.list_teams(org_id)["teams"]) + + role = admin.rbac.create_role(team_id, db, ["read"]) + role_id = (role.get("role") or role)["id"] + assert admin.rbac.get_role(role_id) + admin.rbac.update_role(role_id, permissions=["read", "write"]) + assert any(r["id"] == role_id for r in admin.rbac.list_roles(team_id)["roles"]) + + perm = admin.rbac.create_measurement_permission(role_id, measurement, ["read"]) + perm_id = (perm.get("permission") or perm.get("measurement_permission") or perm)["id"] + assert admin.rbac.list_measurement_permissions(role_id) + + token_id, token = tokens.create([]) + admin.rbac.add_token_to_team(token_id, team_id) + assert str(team_id) in str(admin.rbac.list_token_teams(token_id)) + effective = admin.rbac.get_token_permissions(token_id) + assert db in str(effective), effective + admin.auth.invalidate_cache() + with make_client(token=token) as client: + # The databases routes consult RBAC (internal/api/databases.go + # withResourceReadAuth), so the team grant is honored there. + assert client.databases.get(db).name == db + assert any( + m.name == measurement for m in client.databases.list_measurements(db).measurements + ) + # The SQL routes check only the coarse "read" bit + # (internal/api/query.go withReadAuth): an RBAC-only token cannot query. + with pytest.raises(ArcPermissionError): + client.query.query(f"SELECT count(*) FROM {measurement}", db) + admin.rbac.remove_token_from_team(token_id, team_id) + + admin.rbac.delete_measurement_permission(perm_id) + admin.rbac.delete_role(role_id) + admin.rbac.delete_team(team_id) + with pytest.raises(ArcNotFoundError): + admin.rbac.get_team(team_id) + finally: + with contextlib.suppress(ArcError): + admin.rbac.delete_organization(org_id) + + +@licensed +def test_audit_logs_and_stats(admin: ArcClient) -> None: + marker = uid("audit") + admin.query.query(f"SELECT '{marker}' AS m") + stats = admin.audit.stats() + assert isinstance(stats, dict) + logs = admin.audit.logs(limit=50) + assert isinstance(logs, (dict, list)) + assert logs + + +@licensed +def test_queries_active_history(admin: ArcClient) -> None: + result = admin.query.query("SELECT 7 AS seven") + history = admin.queries.history(limit=50) + assert result.query_id in str(history) + assert isinstance(admin.queries.active(), dict) + with pytest.raises(ArcNotFoundError): + admin.queries.get("no-such-query-id") + + +async def test_async_enterprise_sweep(aadmin: AsyncArcClient) -> None: + if not LICENSED: + pytest.skip("Set ARC_TEST_LICENSED=1") + assert "organizations" in await aadmin.rbac.list_organizations() + assert "policies" in await aadmin.governance.list_policies() + assert isinstance(await aadmin.audit.stats(), dict) + assert isinstance(await aadmin.queries.history(limit=1), dict) + assert (await aadmin.reconciliation.status())["enabled"] is False + assert (await aadmin.cluster.status())["enabled"] is False + assert "status" in await aadmin.backup.status() + + +# -------------------------------------------------------------------------- +# Edge sync +# -------------------------------------------------------------------------- + + +@pytest.fixture +def hub() -> Iterator[ArcClient]: + with make_client(HUB_URL, HUB_TOKEN, timeout=300) as client: + yield client + + +@pytest.fixture +def spoke() -> Iterator[ArcClient]: + with make_client(SPOKE_URL, SPOKE_TOKEN, timeout=300) as client: + yield client + + +@needs_hub +def test_edge_register_spoke_without_name(hub: ArcClient) -> None: + """``register_spoke(spoke_id)`` uses the signature's default ``name=""``. + + Arc requires a non-blank name (internal/edgesync/registry.go Register), so + the default must not produce a request Arc always rejects. + """ + spoke_id = uid("spk").replace("_", "-") + try: + secret = hub.edge_sync.register_spoke(spoke_id) + assert secret.spoke_id == spoke_id and secret.secret + finally: + with contextlib.suppress(ArcError): + hub.edge_sync.delete_spoke(spoke_id, confirm=True) + + +@needs_hub +def test_edge_hub_spoke_registration_lifecycle(hub: ArcClient) -> None: + spoke_id = uid("spk").replace("_", "-") + secret = hub.edge_sync.register_spoke(spoke_id, name="throwaway") + try: + assert secret.spoke_id == spoke_id and len(secret.secret) >= 32 + with pytest.raises(ArcServerError) as dup: + hub.edge_sync.register_spoke(spoke_id, name="again") + assert dup.value.status_code == 409 + + got = hub.edge_sync.get_spoke(spoke_id) + assert got.spoke_id == spoke_id and got.name == "throwaway" and got.enabled + assert spoke_id in {s.spoke_id for s in hub.edge_sync.list_spokes()} + + hub.edge_sync.disable_spoke(spoke_id) + assert hub.edge_sync.get_spoke(spoke_id).enabled is False + hub.edge_sync.enable_spoke(spoke_id) + assert hub.edge_sync.get_spoke(spoke_id).enabled is True + + with pytest.raises(ArcValidationError): + hub.edge_sync.rotate_spoke_secret(spoke_id) + rotated = hub.edge_sync.rotate_spoke_secret(spoke_id, confirm=True) + assert rotated.secret and rotated.secret != secret.secret + + assert isinstance(hub.edge_sync.import_history(spoke_id, limit=5), dict) + finally: + with pytest.raises(ArcValidationError): + hub.edge_sync.delete_spoke(spoke_id) + hub.edge_sync.delete_spoke(spoke_id, confirm=True) + with pytest.raises(ArcNotFoundError): + hub.edge_sync.get_spoke(spoke_id) + + +def _spoke_write(spoke: ArcClient, measurement: str, n: int) -> None: + base = (int(time.time()) - 120) * 1_000_000 + spoke.write.write_columnar( + measurement, + {"time": [base + i for i in range(n)], "v": [float(i) for i in range(n)]}, + database="edge", + ) + server_flush(SPOKE_URL, SPOKE_TOKEN) + + +@needs_spoke +def test_edge_spoke_network_sync(hub: ArcClient, spoke: ArcClient) -> None: + # The spoke discovers new files only inside a sync or export pass + # (internal/edgesync/discover.go), so the ledger is checked afterwards. + before = hub.edge_sync.get_spoke(SPOKE_ID).files_received + synced_before = spoke.edge_sync.spoke_status().get("synced", 0) + _spoke_write(spoke, uid("edge_net"), 3) + result = spoke.edge_sync.run_sync() + assert isinstance(result, dict) + status = spoke.edge_sync.spoke_status() + assert status.get("synced", 0) > synced_before, (result, status) + assert status.get("pending", 0) == 0, status + assert status.get("last_synced_at"), status + assert hub.edge_sync.get_spoke(SPOKE_ID).files_received > before + assert spoke.edge_sync.spoke_ledger(state="synced", limit=5) + with pytest.raises(ArcValidationError): + spoke.edge_sync.dismiss_failed(all_failed=True) + assert "requeued" in spoke.edge_sync.requeue_failed(all_failed=True) + + +@needs_bundle +def test_edge_bundle_export_import_ack(hub: ArcClient, spoke: ArcClient) -> None: + assert BUNDLE_DIR + spoke.edge_sync.run_sync() # drain anything earlier tests left pending + exported_before = spoke.edge_sync.spoke_status().get("exported", 0) + _spoke_write(spoke, uid("edge_bundle"), 2) + out = os.path.join(BUNDLE_DIR, uid("b")) + os.makedirs(out, exist_ok=True) + exported = spoke.edge_sync.export_bundle(out, limit=100) + assert exported.get("exported") is True, exported + bundle_path = exported["dir"] + assert bundle_path.startswith(out) and exported["files"] > 0, exported + assert spoke.edge_sync.spoke_status().get("exported", 0) == exported_before + exported["files"] + + imported = hub.edge_sync.import_bundle(bundle_path) + assert isinstance(imported, dict) and not imported.get("conflicts"), imported + history = hub.edge_sync.import_history(SPOKE_ID, limit=5) + assert exported.get("bundle_id", "") in str(history), history + with pytest.raises(ArcServerError) as again: + hub.edge_sync.import_bundle(bundle_path) + assert again.value.status_code == 409 + + acked = spoke.edge_sync.apply_ack(bundle_path) + assert acked.get("applied") is True, acked + assert spoke.edge_sync.spoke_status().get("exported", 0) == exported_before + + +@needs_bundle +def test_edge_bundle_revert_and_empty_export(spoke: ArcClient) -> None: + assert BUNDLE_DIR + spoke.edge_sync.run_sync() + _spoke_write(spoke, uid("edge_revert"), 1) + out = os.path.join(BUNDLE_DIR, uid("r")) + os.makedirs(out, exist_ok=True) + exported = spoke.edge_sync.export_bundle(out) + assert exported.get("exported") is True, exported + reverted = spoke.edge_sync.revert_export(exported["bundle_id"]) + assert reverted.get("reverted", 0) > 0, reverted + # Drain via the network so the next export is empty. + spoke.edge_sync.run_sync() + empty_dir = os.path.join(BUNDLE_DIR, uid("e")) + os.makedirs(empty_dir, exist_ok=True) + empty = spoke.edge_sync.export_bundle(empty_dir) + assert empty.get("exported") is False, empty + + +@needs_spoke +async def test_async_edge_sync(hub: ArcClient) -> None: + async with make_async_client(HUB_URL, HUB_TOKEN) as ahub: + assert SPOKE_ID in {s.spoke_id for s in await ahub.edge_sync.list_spokes()} + async with make_async_client(SPOKE_URL, SPOKE_TOKEN) as aspoke: + assert "pending" in await aspoke.edge_sync.spoke_status() + + +def test_edge_routes_absent_on_plain_server(admin: ArcClient) -> None: + """Without edge_sync enabled the routes are not mounted: 404, not a crash.""" + with pytest.raises(ArcNotFoundError): + admin.edge_sync.list_spokes() + with pytest.raises(ArcNotFoundError): + admin.edge_sync.spoke_status() + + +# -------------------------------------------------------------------------- +# Telemetry header +# -------------------------------------------------------------------------- + + +def test_installation_id_header_sent(admin: ArcClient) -> None: + counter = RequestCounter(admin) + admin.health() + admin.query.query("SELECT 1") + ids = {r.headers.get(INSTALLATION_HEADER) for r in counter.requests} + assert len(ids) == 1 and None not in ids, ids + uuid.UUID(ids.pop()) + assert "python" in counter.requests[0].headers.get("user-agent", "").lower() + + +@pytest.mark.parametrize("env", [{"DO_NOT_TRACK": "1"}, {"ARC_CLIENT_SEND_INSTALLATION_ID": "0"}]) +def test_installation_id_opt_out(monkeypatch: pytest.MonkeyPatch, env: dict[str, str]) -> None: + for key, value in env.items(): + monkeypatch.setenv(key, value) + with make_client() as client: + counter = RequestCounter(client) + assert client.health().status == "ok" + assert client.query.query("SELECT 1").row_count == 1 + assert all(INSTALLATION_HEADER not in r.headers for r in counter.requests) + + +def test_installation_id_param_opt_out() -> None: + with make_client(send_installation_id=False) as client: + counter = RequestCounter(client) + assert client.health().status == "ok" + assert all(INSTALLATION_HEADER not in r.headers for r in counter.requests) From 3d031af1dc3efd4643fdb116b236938c8b40a11b Mon Sep 17 00:00:00 2001 From: Ignacio Van Droogenbroeck <64545348+xe-nvdk@users.noreply.github.com> Date: Mon, 2 Nov 2026 10:53:42 -0600 Subject: [PATCH 02/10] test(integration): cover async delete, deleted-database listing, bundle cleanup --- tests/integration/test_live_parity.py | 28 +++++++++++++++++++++++++++ 1 file changed, 28 insertions(+) diff --git a/tests/integration/test_live_parity.py b/tests/integration/test_live_parity.py index e9f322b..8adc287 100644 --- a/tests/integration/test_live_parity.py +++ b/tests/integration/test_live_parity.py @@ -49,6 +49,7 @@ import contextlib import glob import os +import shutil import threading import time import uuid @@ -835,11 +836,35 @@ def test_databases_delete_allow_unreachable(admin: ArcClient) -> None: admin.databases.get(name) +def test_deleted_database_not_listed(admin: ArcClient) -> None: + """After ``databases.delete`` the database must not be listed any more. + + On local storage Arc lists databases from directories + (internal/api/databases.go listDatabases), and a delete removes the files + but leaves the empty directory tree, so ``list()`` keeps showing a + database that ``get()`` answers 404 for. + """ + name = uid("sdk_dblist") + admin.databases.create(name) + admin.write.write_columnar("m", {"time": [int(time.time() * 1e6)], "v": [1.0]}, database=name) + server_flush() + admin.databases.delete(name, confirm=True) + with pytest.raises(ArcNotFoundError): + admin.databases.get(name) + listed = [d for d in admin.databases.list().databases if d.name == name] + assert not listed, f"deleted database still listed: {listed}" + + async def test_async_databases(aadmin: AsyncArcClient, seeded: tuple[str, str, int]) -> None: db, measurement, _ = seeded schema = await aadmin.databases.get_measurement_schema(db, measurement) assert any(f.name == "usage" for f in schema.fields) assert (await aadmin.databases.get(db)).name == db + preview = await aadmin.delete.delete( + db, measurement, "host = 'b'", dry_run=True, allow_unreachable=True + ) + assert preview.dry_run is True and preview.deleted_count == 1 + assert (await aadmin.delete.get_config()).enabled is True # -------------------------------------------------------------------------- @@ -1436,6 +1461,7 @@ def test_edge_bundle_export_import_ack(hub: ArcClient, spoke: ArcClient) -> None acked = spoke.edge_sync.apply_ack(bundle_path) assert acked.get("applied") is True, acked assert spoke.edge_sync.spoke_status().get("exported", 0) == exported_before + shutil.rmtree(out, ignore_errors=True) @needs_bundle @@ -1455,6 +1481,8 @@ def test_edge_bundle_revert_and_empty_export(spoke: ArcClient) -> None: os.makedirs(empty_dir, exist_ok=True) empty = spoke.edge_sync.export_bundle(empty_dir) assert empty.get("exported") is False, empty + shutil.rmtree(out, ignore_errors=True) + shutil.rmtree(empty_dir, ignore_errors=True) @needs_spoke From 49ec136d8bb90ea670bce91bdd035e99a9aa183d Mon Sep 17 00:00:00 2001 From: Ignacio Van Droogenbroeck <64545348+xe-nvdk@users.noreply.github.com> Date: Mon, 2 Nov 2026 10:53:53 -0600 Subject: [PATCH 03/10] test(integration): add live cluster suite for the 0.2.0 cluster surface --- tests/integration/test_live_cluster.py | 1068 ++++++++++++++++++++++++ 1 file changed, 1068 insertions(+) create mode 100644 tests/integration/test_live_cluster.py diff --git a/tests/integration/test_live_cluster.py b/tests/integration/test_live_cluster.py new file mode 100644 index 0000000..d6d01d8 --- /dev/null +++ b/tests/integration/test_live_cluster.py @@ -0,0 +1,1068 @@ +"""Opt-in live tests of the cluster-only SDK surface against a real Arc cluster. + +Skipped unless ``ARC_TEST_CLUSTER_URLS`` is set: a comma-separated list of the +direct URL of every node (not a load balancer), plus ``ARC_TEST_TOKEN``, an +admin token valid on all of them. Written against a 4-node local-storage +Arc Enterprise cluster (three writers, one reader, peer replication, delete +and reconciliation enabled) built from Arc main (2027.01.1):: + + N=http://localhost + ARC_TEST_CLUSTER_URLS=$N:8001,$N:8002,$N:8003,$N:8004 ARC_TEST_TOKEN=... \\ + uv run pytest tests/integration/test_live_cluster.py -v + +The suite is DESTRUCTIVE: it deletes cluster files, demotes the primary +writer, converges the Raft voter set and restores backups. Use a throwaway +cluster. Tests run in file order and each restores what it changes (the +primary writer is handed back to the node that held it). + +Tests that simulate node faults (a stopped writer, a manifest file missing +from the primary's disk) drive the containers with ``docker`` and assume +each container is named after its Arc node id; they are skipped when the +``docker`` CLI is not available. ``ARC_TEST_CLUSTER_DATA_DIR`` overrides the +in-container storage root (default ``/app/data/arc``). + +``remove_node`` takes a node out of Raft for good, so it runs only with +``ARC_TEST_CLUSTER_REMOVE_NODE=1``; it removes the reader and then restarts +its container, which may leave the cluster needing a rebuild. +""" + +from __future__ import annotations + +import contextlib +import os +import shutil +import subprocess +import time +import uuid +from collections.abc import Iterator +from dataclasses import dataclass, field +from typing import Any, Callable, Optional, TypeVar +from urllib.parse import urlparse + +import httpx +import pytest + +from arc_client import ArcClient, AsyncArcClient +from arc_client.exceptions import ( + ArcConnectionError, + ArcIngestRefusedError, + ArcNotFoundError, + ArcServerError, + ArcValidationError, +) + +URLS = [u.strip() for u in os.environ.get("ARC_TEST_CLUSTER_URLS", "").split(",") if u.strip()] +TOKEN = os.environ.get("ARC_TEST_TOKEN") +DATA_DIR = os.environ.get("ARC_TEST_CLUSTER_DATA_DIR", "/app/data/arc") +HAVE_DOCKER = shutil.which("docker") is not None + +pytestmark = pytest.mark.skipif( + not URLS, + reason="Set ARC_TEST_CLUSTER_URLS (and ARC_TEST_TOKEN) to run live cluster tests.", +) +needs_docker = pytest.mark.skipif(not HAVE_DOCKER, reason="needs the docker CLI") + +T = TypeVar("T") + +# Arc's delete waits this long for manifest files missing on the node +# (internal/api/delete_records.go deleteLocalWait). +ARC_DELETE_LOCAL_WAIT = 30.0 + + +# -------------------------------------------------------------------------- +# Helpers +# -------------------------------------------------------------------------- + + +def _client_for(url: str, **kwargs: Any) -> ArcClient: + parsed = urlparse(url) + return ArcClient( + host=parsed.hostname or "localhost", + port=parsed.port or (443 if parsed.scheme == "https" else 80), + ssl=parsed.scheme == "https", + token=TOKEN, + **kwargs, + ) + + +def _async_client_for(url: str, **kwargs: Any) -> AsyncArcClient: + parsed = urlparse(url) + return AsyncArcClient( + host=parsed.hostname or "localhost", + port=parsed.port or (443 if parsed.scheme == "https" else 80), + ssl=parsed.scheme == "https", + token=TOKEN, + **kwargs, + ) + + +def wait_for( + fn: Callable[[], T], + until: Callable[[T], bool], + timeout: float = 30.0, + interval: float = 0.5, + what: str = "condition", +) -> T: + """Poll ``fn`` until ``until(result)``; fail with the last result on timeout.""" + deadline = time.monotonic() + timeout + last: Any = None + last_exc: Optional[BaseException] = None + while time.monotonic() < deadline: + try: + last = fn() + last_exc = None + if until(last): + return last + except Exception as e: # node restarting, leadership moving + last_exc = e + time.sleep(interval) + pytest.fail(f"timed out waiting for {what}; last={last!r} exc={last_exc!r}") + + +def docker(*args: str, check: bool = True) -> str: + out = subprocess.run(["docker", *args], capture_output=True, text=True, timeout=120) + if check and out.returncode != 0: + raise RuntimeError(f"docker {' '.join(args)} failed: {out.stderr.strip()}") + return out.stdout + + +def exists_on(container: str, path: str) -> bool: + """Whether ``path`` (a manifest path) is on the container's local storage.""" + rc = subprocess.run( + ["docker", "exec", container, "test", "-e", f"{DATA_DIR}/{path}"], + capture_output=True, + timeout=60, + ).returncode + return rc == 0 + + +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") + + +def unique(prefix: str) -> str: + return f"{prefix}_{uuid.uuid4().hex[:8]}" + + +@dataclass +class Cluster: + """Node-id keyed clients and dynamic leader / primary lookup.""" + + urls: list[str] + clients: dict[str, ArcClient] = field(default_factory=dict) + url_of: dict[str, str] = field(default_factory=dict) + + def any(self) -> ArcClient: + for client in self.clients.values(): + with contextlib.suppress(Exception): + client.cluster.health() + return client + raise RuntimeError("no cluster node answers") + + def status(self) -> dict[str, Any]: + # Ask the leader-agnostic status of the first node that answers. + return self.any().cluster.status() + + def leader_id(self) -> str: + return self.status()["raft"]["leader_id"] + + def primary_id(self) -> str: + return self.status()["primary_writer_id"] + + def leader(self) -> ArcClient: + return self.clients[self.leader_id()] + + def follower(self) -> ArcClient: + leader = self.leader_id() + return next(c for nid, c in self.clients.items() if nid != leader) + + def role_of(self, node_id: str) -> str: + return self.clients[node_id].cluster.local()["role"] + + def writers(self) -> list[str]: + return [nid for nid in self.clients if self.role_of(nid) == "writer"] + + def readers(self) -> list[str]: + return [nid for nid in self.clients if self.role_of(nid) == "reader"] + + def standby_writer(self, exclude: tuple[str, ...] = ()) -> str: + primary = self.primary_id() + leader = self.leader_id() + candidates = [w for w in self.writers() if w != primary and w not in exclude] + # Prefer a standby that is not the Raft leader either. + candidates.sort(key=lambda w: w == leader) + if not candidates: + pytest.skip("cluster has no standby writer") + return candidates[0] + + def wait_settled(self, timeout: float = 90.0) -> dict[str, Any]: + """Wait for every node healthy, a leader, a primary, and no failover running.""" + n = len(self.clients) + + def settled(s: dict[str, Any]) -> bool: + return ( + s.get("healthy_count") == n + and bool(s.get("primary_writer_id")) + and bool(s.get("raft", {}).get("leader_id")) + and not s.get("writer_failover", {}).get("failover_in_progress", False) + ) + + return wait_for( + self.status, settled, timeout=timeout, interval=1.0, what="cluster to settle" + ) + + +@pytest.fixture(scope="module") +def cluster() -> Iterator[Cluster]: + c = Cluster(URLS) + for url in URLS: + client = _client_for(url) + node_id = client.cluster.local()["id"] + c.clients[node_id] = client + c.url_of[node_id] = url + yield c + for client in c.clients.values(): + client.close() + + +@pytest.fixture +def dead() -> Iterator[ArcClient]: + """A client pointed at a closed port: any request raises ArcConnectionError.""" + client = ArcClient(host="127.0.0.1", port=1, token="x", max_retries=0, timeout=2) + yield client + client.close() + + +def write_and_locate( + cluster: Cluster, database: str, measurement: str, lines: list[str] +) -> list[dict[str, Any]]: + """Write via the primary, flush it, and wait for the manifest to list the files.""" + primary = cluster.clients[cluster.primary_id()] + primary.write.write_line_protocol(lines, database=database) + flush(primary) + return wait_for( + lambda: [ + f + for f in cluster.leader().cluster.files(database=database)["files"] + if f["measurement"] == measurement + ], + bool, + what=f"manifest to list {database}.{measurement}", + ) + + +# -------------------------------------------------------------------------- +# Read routes +# -------------------------------------------------------------------------- + + +def test_status_on_every_node(cluster: Cluster) -> None: + leader_ids = set() + for node_id, client in cluster.clients.items(): + status = client.cluster.status() + assert status["enabled"] is True + assert status["mode"] == "cluster" + assert status["local_node_id"] == node_id + assert status["node_count"] == len(cluster.clients) + assert status["license"]["valid"] is True + assert "clustering" in status["license"]["features"] + leader_ids.add(status["raft"]["leader_id"]) + # Every node agrees on who leads. + assert len(leader_ids) == 1 + + +def test_nodes_node_local_health(cluster: Cluster) -> None: + ids = set(cluster.clients) + for node_id, client in cluster.clients.items(): + listing = client.cluster.nodes() + assert listing["total"] == len(ids) + assert {n["id"] for n in listing["nodes"]} == ids + + local = client.cluster.local() + assert local["id"] == node_id + assert local["is_local"] is True + assert set(local["capabilities"]) == { + "can_ingest", + "can_query", + "can_compact", + "can_coordinate", + } + + for other in ids: + assert client.cluster.node(other)["id"] == other + + health = client.cluster.health() + assert health["total"] == len(ids) + assert health["healthy"] == len(ids) + assert health["unhealthy"] == 0 + + # Exactly one writer reports itself primary, and it matches status(). + primaries = [ + n["id"] for n in cluster.any().cluster.nodes()["nodes"] if n["writer_state"] == "primary" + ] + assert primaries == [cluster.primary_id()] + + +def test_nodes_filters(cluster: Cluster) -> None: + client = cluster.any() + readers = client.cluster.nodes(role="reader") + assert readers["total"] == len(cluster.readers()) + assert all(n["role"] == "reader" for n in readers["nodes"]) + assert client.cluster.nodes(role="writer", state="healthy")["total"] == len(cluster.writers()) + with pytest.raises(ArcServerError) as exc: + client.cluster.nodes(role="bogus") + assert exc.value.status_code == 400 + + +def test_node_unknown_raises_not_found(cluster: Cluster) -> None: + with pytest.raises(ArcNotFoundError): + cluster.any().cluster.node("no-such-node") + + +def test_files_pagination_shape(cluster: Cluster) -> None: + db = unique("sdkcl_pg") + write_and_locate(cluster, db, "pg", ["pg,host=a v=1", "pg2,host=a v=2"]) + wait_for( + lambda: cluster.leader().cluster.files(database=db)["total"], + lambda t: t >= 2, + what="both files in the manifest", + ) + page = cluster.leader().cluster.files(database=db, limit=1) + assert page["total"] == 1 + assert page["next_cursor"] + rest = cluster.leader().cluster.files(database=db, cursor=page["next_cursor"], limit=10) + assert {f["path"] for f in rest["files"]}.isdisjoint({f["path"] for f in page["files"]}) + assert rest["next_cursor"] == "" + + +async def test_async_cluster_reads_and_errors(cluster: Cluster) -> None: + leader_id = cluster.leader_id() + follower_id = next(n for n in cluster.clients if n != leader_id) + async with _async_client_for(cluster.url_of[leader_id]) as leader: + status = await leader.cluster.status() + assert status["raft"]["leader_id"] == leader_id + assert (await leader.cluster.local())["id"] == leader_id + assert (await leader.cluster.nodes())["total"] == len(cluster.clients) + assert (await leader.cluster.node(follower_id))["id"] == follower_id + assert (await leader.cluster.health())["healthy"] == len(cluster.clients) + assert "files" in await leader.cluster.files(limit=1) + claims = await leader.cluster.stream_claims() + assert isinstance(claims["claims"], list) + with pytest.raises(ArcNotFoundError): + await leader.cluster.abort_compaction_job("sdk-no-such-job", confirm=True) + plan = await leader.cluster.converge_voters() + assert plan["success"] is True + assert plan["result"]["dry_run"] is True + async with _async_client_for(cluster.url_of[follower_id]) as follower: + with pytest.raises(ArcServerError) as exc: + await follower.cluster.policies() + assert exc.value.status_code == 409 + assert exc.value.response_body["leader"] == leader_id + + +# -------------------------------------------------------------------------- +# Readiness +# -------------------------------------------------------------------------- + + +def test_write_readiness_by_role(cluster: Cluster) -> None: + primary = cluster.primary_id() + for node_id, client in cluster.clients.items(): + ready = client.write_readiness() + if node_id == primary: + assert ready.ready is True + assert ready.status == "ready" + else: + # Readers and standby writers in local-storage mode. + assert ready.ready is False, node_id + assert ready.status == "not_write_target", node_id + assert ready.reason + # General readiness is unaffected by the write role. + assert client.readiness().ready is True + + +# -------------------------------------------------------------------------- +# confirm=False gates: nothing may be sent (a dead endpoint proves it) +# -------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "call", + [ + pytest.param(lambda c: c.cluster.delete_file("db/m/x.parquet"), id="delete_file"), + pytest.param(lambda c: c.cluster.demote_writer("arc-writer1"), id="demote_writer"), + pytest.param(lambda c: c.cluster.converge_voters(dry_run=False), id="converge_act"), + pytest.param( + lambda c: c.cluster.release_stream_claim("00000000000000ab", 1, abandon=True), + id="release_abandon", + ), + pytest.param(lambda c: c.cluster.abort_compaction_job("job"), id="abort_compaction"), + pytest.param( + lambda c: c.cluster.adopt_policies("retention", "arc-writer1", "h"), + id="adopt_policies", + ), + pytest.param(lambda c: c.backup.restore("backup-x"), id="backup_restore"), + pytest.param(lambda c: c.databases.delete("somedb"), id="databases_delete"), + pytest.param(lambda c: c.reconciliation.trigger(dry_run=False), id="reconcile_act"), + ], +) +def test_confirm_gates_send_nothing(dead: ArcClient, call: Callable[[ArcClient], Any]) -> None: + with pytest.raises(ArcValidationError): + call(dead) + + +def test_dead_client_really_sends(dead: ArcClient) -> None: + # Control for the gate test: the same client does reach the network. + with pytest.raises(ArcConnectionError): + dead.cluster.delete_file("db/m/x.parquet", confirm=True) + + +# -------------------------------------------------------------------------- +# delete_file +# -------------------------------------------------------------------------- + + +def test_delete_file_removes_it_cluster_wide(cluster: Cluster) -> None: + db = unique("sdkcl_df") + files = write_and_locate(cluster, db, "df", ["df,host=a v=1", "df,host=b v=2"]) + path = files[0]["path"] + # Wait for every node's manifest to carry it first, so "gone" means deleted. + for client in cluster.clients.values(): + wait_for( + lambda c=client: {f["path"] for f in c.cluster.files(database=db)["files"]}, + lambda paths: path in paths, + what=f"{path} on every node", + ) + if HAVE_DOCKER: + for node_id in cluster.clients: + wait_for( + lambda n=node_id: exists_on(n, path), bool, what=f"replicated copy on {node_id}" + ) + + result = cluster.leader().cluster.delete_file(path, confirm=True, reason="sdk live test") + assert result["success"] is True + assert result["path"] == path + + for client in cluster.clients.values(): + wait_for( + lambda c=client: {f["path"] for f in c.cluster.files(database=db)["files"]}, + lambda paths: path not in paths, + what=f"{path} gone from every manifest", + ) + if HAVE_DOCKER: + for node_id in cluster.clients: + wait_for( + lambda n=node_id: exists_on(n, path), + lambda present: not present, + what=f"{path} unlinked on {node_id}", + ) + + # A second delete of the same path is a 404. + with pytest.raises(ArcNotFoundError): + cluster.leader().cluster.delete_file(path, confirm=True) + + +def test_delete_file_from_follower(cluster: Cluster) -> None: + """The manifest delete goes through Raft; a follower must still accept it.""" + db = unique("sdkcl_dff") + path = write_and_locate(cluster, db, "dff", ["dff,host=a v=1"])[0]["path"] + follower = cluster.follower() + wait_for( + lambda: {f["path"] for f in follower.cluster.files(database=db)["files"]}, + lambda paths: path in paths, + what="follower manifest", + ) + result = follower.cluster.delete_file(path, confirm=True) + assert result["success"] is True + wait_for( + lambda: {f["path"] for f in cluster.leader().cluster.files(database=db)["files"]}, + lambda paths: path not in paths, + what="gone on the leader", + ) + + +def test_delete_file_unknown_path(cluster: Cluster) -> None: + with pytest.raises(ArcNotFoundError): + cluster.leader().cluster.delete_file("sdkcl_none/m/2020/01/01/00/x.parquet", confirm=True) + + +# -------------------------------------------------------------------------- +# Reconciliation +# -------------------------------------------------------------------------- + + +def test_reconciliation_status_trigger_get_run(cluster: Cluster) -> None: + for node_id, client in cluster.clients.items(): + status = client.reconciliation.status() + assert status["enabled"] is True, node_id + run = client.reconciliation.trigger() + assert run.dry_run is True + assert run.role == cluster.role_of(node_id) + # Runs live in the node's own in-memory buffer. + fetched = client.reconciliation.get_run(run.id) + assert fetched.id == run.id + assert fetched.dry_run is True + with pytest.raises(ArcNotFoundError): + cluster.leader().reconciliation.get_run("sdk-no-such-run") + + +def test_reconciliation_act_when_dry_run_is_clean(cluster: Cluster) -> None: + leader = cluster.leader() + plan = leader.reconciliation.trigger() + if plan.manifest_deletes or plan.storage_deletes or plan.orphan_manifest_count: + pytest.skip(f"dry run would delete something: {plan.model_dump()}") + run = leader.reconciliation.trigger(dry_run=False, confirm=True) + assert run.dry_run is False + assert run.manifest_deletes == 0 + assert run.storage_deletes == 0 + assert leader.reconciliation.get_run(run.id).dry_run is False + + +# -------------------------------------------------------------------------- +# Data DELETE routing and allow_unreachable +# -------------------------------------------------------------------------- + + +def test_delete_on_standby_writer_is_refused_not_forwarded(cluster: Cluster) -> None: + """Writes are forwarded to the primary; a data DELETE is not (503, no retry).""" + db = unique("sdkcl_sb") + write_and_locate(cluster, db, "sb", ["sb,host=a v=1"]) + standby = cluster.clients[cluster.standby_writer()] + # A dry run is allowed anywhere. + assert standby.delete.delete(db, "sb", "host = 'a'").success is True + with pytest.raises(ArcServerError) as exc: + standby.delete.delete(db, "sb", "host = 'a'", dry_run=False, confirm=True) + assert exc.value.status_code == 503 + assert "not primary writer" in str(exc.value) + assert not isinstance(exc.value, ArcIngestRefusedError) + + +@pytest.fixture +def missing_on_primary(cluster: Cluster) -> Iterator[tuple[str, str, str]]: + """A manifest file removed from the primary's disk (kept on the other nodes). + + Yields ``(database, measurement, path)``. Teardown removes the path from + the manifest cluster-wide with delete_file. + """ + if not HAVE_DOCKER: + pytest.skip("needs the docker CLI") + db = unique("sdkcl_unr") + path = write_and_locate(cluster, db, "unr", ["unr,host=a v=1", "unr,host=b v=2"])[0]["path"] + primary = cluster.primary_id() + docker("exec", primary, "rm", "-f", f"{DATA_DIR}/{path}") + assert not exists_on(primary, path) + yield db, "unr", path + with contextlib.suppress(ArcNotFoundError): + cluster.leader().cluster.delete_file(path, confirm=True, reason="sdk test cleanup") + + +def _long_timeout_primary(cluster: Cluster) -> ArcClient: + return _client_for( + cluster.url_of[cluster.primary_id()], timeout=ARC_DELETE_LOCAL_WAIT * 3, max_retries=0 + ) + + +@needs_docker +def test_delete_reports_unreachable_files( + cluster: Cluster, missing_on_primary: tuple[str, str, str] +) -> None: + db, measurement, path = missing_on_primary + with _long_timeout_primary(cluster) as primary: + plan = primary.delete.delete(db, measurement, "host = 'a'") + assert plan.dry_run is True + assert plan.unreachable_files == [path] + + # Without the flag: Arc waits deleteLocalWait, then refuses with 503 + # naming the files. The SDK must not retry and must keep the body. + with pytest.raises(ArcServerError) as exc: + primary.delete.delete(db, measurement, "host = 'a'", dry_run=False, confirm=True) + assert exc.value.status_code == 503 + assert exc.value.response_body["unreachable_files"] == [path] + + result = primary.delete.delete( + db, measurement, "host = 'a'", dry_run=False, confirm=True, allow_unreachable=True + ) + assert result.success is True + assert result.unreachable_files == [path] + assert result.unflushed_nodes == [] # local storage: no peer flush + + +@needs_docker +def test_delete_allow_unreachable_with_default_timeout( + cluster: Cluster, missing_on_primary: tuple[str, str, str] +) -> None: + """The documented allow_unreachable path, with the client's default settings. + + Arc waits up to 30 s for the missing files before it answers, which is + also the SDK's default request timeout. + """ + db, measurement, path = missing_on_primary + with _client_for(cluster.url_of[cluster.primary_id()]) as primary: + result = primary.delete.delete( + db, measurement, "host = 'a'", dry_run=False, confirm=True, allow_unreachable=True + ) + assert result.unreachable_files == [path] + + +@needs_docker +def test_delete_with_a_stopped_writer(cluster: Cluster) -> None: + """On local storage a stopped standby is not "unreachable" for a delete.""" + db = unique("sdkcl_stop") + # host=c stays, so the database still has data when it is dropped below. + write_and_locate(cluster, db, "st", ["st,host=a v=1", "st,host=b v=2", "st,host=c v=3"]) + victim = cluster.standby_writer(exclude=(cluster.leader_id(),)) + primary = cluster.clients[cluster.primary_id()] + docker("stop", victim) + try: + result = primary.delete.delete(db, "st", "host = 'a'", dry_run=False, confirm=True) + assert result.success is True + assert result.deleted_count == 1 + assert result.unreachable_files == [] + assert result.unflushed_nodes == [] + + flagged = primary.delete.delete( + db, "st", "host = 'b'", dry_run=False, confirm=True, allow_unreachable=True + ) + assert flagged.success is True + assert flagged.unreachable_files == [] + + body = primary.databases.delete(db, confirm=True, allow_unreachable=True) + assert "deleted" in body["message"] + assert not body.get("unflushed_nodes") + finally: + docker("start", victim) + cluster.wait_settled() + + +# -------------------------------------------------------------------------- +# Backup / restore +# -------------------------------------------------------------------------- + + +def test_scoped_backup_and_restore_modes(cluster: Cluster) -> None: + db = unique("sdkcl_bk") + write_and_locate(cluster, db, "bk", ["bk,host=a v=1"]) + node = cluster.clients[cluster.primary_id()] + started = node.backup.create(databases=[db]) + assert started["status"] == "running" + assert started["databases"] == [db] + + def latest() -> dict[str, Any]: + return node.backup.status() + + done = wait_for( + latest, + lambda s: s.get("operation") == "backup" and s.get("status") == "completed", + what="backup", + ) + assert done["scope"] == [db] + backup_id = done["backup_id"] + try: + listing = node.backup.list() + assert any(b["backup_id"] == backup_id for b in listing["backups"]) + assert node.backup.get(backup_id) + + for mode in ("replace", "merge"): + restored = node.backup.restore(backup_id, confirm=True, mode=mode) + assert restored["mode"] == mode + final = wait_for( + latest, + lambda s, m=mode: s.get("operation") == "restore" + and s.get("mode") == m + and s.get("status") in ("completed", "failed"), + what=f"restore {mode}", + ) + assert final["status"] == "completed", final + # Arc's default for a scoped backup on a cluster node is replace. + default = node.backup.restore(backup_id, confirm=True) + assert default["mode"] == "replace" + wait_for(latest, lambda s: s.get("status") == "completed", what="default restore") + # The restored data is in the manifest again. + assert cluster.leader().cluster.files(database=db)["total"] >= 1 + finally: + with contextlib.suppress(Exception): + node.backup.delete(backup_id) + + +# -------------------------------------------------------------------------- +# Policies +# -------------------------------------------------------------------------- + + +@pytest.fixture +def policies_present(cluster: Cluster) -> Iterator[None]: + """A retention policy and a CQ, so there is something to adopt.""" + db = unique("sdkcl_pol") + primary = cluster.clients[cluster.primary_id()] + write_and_locate(cluster, db, "src", ["src,host=a v=1"]) + rp = primary.retention.create(unique("rp"), db, retention_days=30) + cq = None + # continuous_query.enabled=false: Arc registers no CQ routes at all + # ("Cannot POST /api/v1/continuous_queries" -> ArcNotFoundError). The + # policies endpoint still reports the cq kind, so the tests go on without one. + with contextlib.suppress(ArcNotFoundError): + cq = primary.continuous_queries.create( + name=unique("cq"), + database=db, + source_measurement="src", + destination_measurement="dst", + query="SELECT time_bucket(INTERVAL '1 minute', time) AS time, avg(v) AS v " + "FROM src WHERE time >= {start_time} AND time < {end_time} GROUP BY 1", + interval="1m", + ) + yield + with contextlib.suppress(Exception): + primary.retention.delete(rp.id) + if cq is not None: + with contextlib.suppress(Exception): + primary.continuous_queries.delete(cq.id) + + +def test_policies_on_leader(cluster: Cluster, policies_present: None) -> None: + leader = cluster.leader() + both = leader.cluster.policies() + assert both["success"] is True + for kind in ("retention", "cq"): + one = leader.cluster.policies(kind=kind) + assert one["success"] is True + assert kind in one + other = "cq" if kind == "retention" else "retention" + assert other not in one + candidates = one[kind] + if candidates["seeded"]: + # The cluster already holds its policies in Raft: there is no + # per-node candidate list to choose from. + assert "nodes" not in candidates + else: + assert candidates["nodes"], candidates + for node_set in candidates["nodes"].values(): + assert node_set.get("hash") or node_set.get("error") + + +def test_policies_on_follower_names_the_leader(cluster: Cluster) -> None: + leader_id = cluster.leader_id() + with pytest.raises(ArcServerError) as exc: + cluster.follower().cluster.policies(kind="retention") + assert exc.value.status_code == 409 + assert exc.value.response_body["leader"] == leader_id + assert "leader" in str(exc.value).lower() + + +def test_policies_invalid_kind_is_local() -> None: + client = ArcClient(host="127.0.0.1", port=1, max_retries=0) + with pytest.raises(ArcValidationError): + client.cluster.policies(kind="tiering") + + +def test_adopt_policies(cluster: Cluster, policies_present: None) -> None: + leader = cluster.leader() + for kind in ("retention", "cq"): + candidates = leader.cluster.policies(kind=kind)[kind] + if not candidates["seeded"]: + node, node_set = next( + (n, s) for n, s in candidates["nodes"].items() if s.get("hash") is not None + ) + accepted = leader.cluster.adopt_policies(kind, node, node_set["hash"], confirm=True) + assert accepted["success"] is True + continue + # Already seeded: Arc refuses with 409, which the SDK raises (the + # 200/success:false path _check_adopted handles is not reached). + with pytest.raises(ArcServerError) as exc: + leader.cluster.adopt_policies(kind, cluster.leader_id(), "deadbeef", confirm=True) + assert exc.value.status_code == 409 + assert "already holds" in str(exc.value) + + +# -------------------------------------------------------------------------- +# Stream claims and wedged compaction jobs +# -------------------------------------------------------------------------- + + +def test_stream_claims_listing(cluster: Cluster) -> None: + for client in cluster.clients.values(): + claims = client.cluster.stream_claims() + for key in ("enabled", "instances", "claims", "departed_writers", "delete_records"): + assert key in claims + status = cluster.status() + assert status["stream_claims"]["claims"] == len( + cluster.leader().cluster.stream_claims()["claims"] + ) + + +def test_release_stream_claim(cluster: Cluster) -> None: + leader = cluster.leader() + claims = leader.cluster.stream_claims()["claims"] + unfinished = [c for c in claims if not c.get("finished")] + if unfinished: # pragma: no cover - not reachable without stream takeover + claim = unfinished[0] + result = leader.cluster.release_stream_claim(claim["instance"], claim["epoch"]) + assert result["success"] is True + return + # No claim exists (stream takeover off): Arc answers 404 on the leader... + with pytest.raises(ArcNotFoundError): + leader.cluster.release_stream_claim(0xAB, 1) + with pytest.raises(ArcNotFoundError): + leader.cluster.release_stream_claim("00000000000000ab", 1, abandon=True, confirm=True) + # ...and 409 naming the leader on a follower. + with pytest.raises(ArcServerError) as exc: + cluster.follower().cluster.release_stream_claim(0xAB, 1) + assert exc.value.status_code == 409 + assert "leader" in str(exc.value) + + +def test_abort_compaction_job_unknown(cluster: Cluster) -> None: + status = cluster.status() + wedged = status.get("compaction_commit", {}).get("wedged_jobs") or [] + assert not wedged or isinstance(wedged, list) + with pytest.raises(ArcNotFoundError): + cluster.leader().cluster.abort_compaction_job("sdk-no-such-job", confirm=True) + # Path segments are escaped, so a job id with a slash cannot change the route. + with pytest.raises(ArcNotFoundError): + cluster.leader().cluster.abort_compaction_job("a/b", confirm=True) + + +# -------------------------------------------------------------------------- +# Compactor lease +# -------------------------------------------------------------------------- + + +def test_assign_compactor(cluster: Cluster) -> None: + leader = cluster.leader() + nodes = leader.cluster.nodes()["nodes"] + holder = next((n["id"] for n in nodes if n["is_active_compactor"]), None) + has_compactor_role = any(n["role"] == "compactor" for n in nodes) + target = next(n["id"] for n in nodes if n["role"] == "writer" and n["id"] != holder) + + # Non-leader: always 409. + follower = cluster.follower() + with pytest.raises(ArcServerError) as exc: + follower.cluster.assign_compactor(target) + assert exc.value.status_code == 409 + + if not has_compactor_role and holder is None: + # No compactor node: nothing manages a lease, and Arc refuses before + # it even looks at the target (so an unknown node is a 409, not 404). + for node_id in (target, cluster.readers()[0] if cluster.readers() else target, "nope"): + with pytest.raises(ArcServerError) as exc: + leader.cluster.assign_compactor(node_id) + assert exc.value.status_code == 409 + assert "does not manage a compactor lease" in str(exc.value) + assert cluster.status()["active_compactor"] == {"assigned": False} + return + + result = leader.cluster.assign_compactor(target) # pragma: no cover - topology dependent + assert result["success"] is True + assert result["node_id"] == target + assert "preemption_suppressed_seconds" in result + wait_for( + lambda: [n["id"] for n in leader.cluster.nodes()["nodes"] if n["is_active_compactor"]], + lambda ids: ids == [target], + what="compactor lease to move", + ) + assert cluster.status()["active_compactor"]["node_id"] == target + + +# -------------------------------------------------------------------------- +# Raft voter set +# -------------------------------------------------------------------------- + + +def test_converge_voters(cluster: Cluster) -> None: + leader = cluster.leader() + plan = leader.cluster.converge_voters() + assert plan["success"] is True + result = plan["result"] + assert result["dry_run"] is True + before = result["voting_servers_before"] + + with pytest.raises(ArcServerError) as exc: + cluster.follower().cluster.converge_voters() + assert exc.value.status_code == 409 + + if result.get("demoted") or result["voting_servers_after"] != before: + pytest.skip(f"converge would change the voter set: {result}") + acted = leader.cluster.converge_voters(dry_run=False, confirm=True) + assert acted["success"] is True + assert acted["result"]["dry_run"] is False + assert acted["result"]["voting_servers_after"] == before + assert not acted["result"].get("demoted") + membership = cluster.status()["raft"]["membership"] + assert membership["mismatches"] == 0 + assert membership["voting_servers"] == before + + +# -------------------------------------------------------------------------- +# Fenced writes (ingest refusals) +# -------------------------------------------------------------------------- + +# A standby writer forwards writes to the primary unless the request says it +# was already forwarded; then it handles the write locally and refuses it with +# 503 + X-Arc-Ingest-Fenced, exactly as during a failover window. +FORWARDED_BY = "X-Arc-Forwarded-By" + + +def _fenced_client(cluster: Cluster, fence_first_only: bool, max_retries: int) -> Any: + standby = cluster.standby_writer() + client = _client_for(cluster.url_of[standby], max_retries=max_retries, database="sdkcl_fence") + sent: list[int] = [] + + def hook(request: httpx.Request) -> None: + if "/write/" not in request.url.path: + return + sent.append(1) + if not fence_first_only or len(sent) == 1: + request.headers[FORWARDED_BY] = "sdk-live-test" + + client._get_http()._get_client().event_hooks["request"] = [hook] + return client, sent + + +def test_fenced_single_measurement_write_is_resent(cluster: Cluster) -> None: + client, sent = _fenced_client(cluster, fence_first_only=True, max_retries=2) + with client: + client.write.write_line_protocol(["fence_one,host=a v=1"]) + # First attempt fenced, the resend (after Retry-After) forwarded and stored. + assert len(sent) == 2 + + +def test_fenced_single_measurement_retries_then_raises(cluster: Cluster) -> None: + client, sent = _fenced_client(cluster, fence_first_only=False, max_retries=1) + with client, pytest.raises(ArcIngestRefusedError) as exc: + client.write.write_line_protocol(["fence_one,host=a v=1"]) + assert exc.value.reason == "fenced" + assert exc.value.retry_safe is True + assert exc.value.status_code == 503 + assert len(sent) == 2 + + +def test_fenced_multi_measurement_line_protocol_not_resent(cluster: Cluster) -> None: + client, sent = _fenced_client(cluster, fence_first_only=False, max_retries=3) + with client, pytest.raises(ArcIngestRefusedError) as exc: + client.write.write_line_protocol(["fence_a,host=a v=1", "fence_b,host=a v=2"]) + assert exc.value.reason == "fenced" + assert exc.value.retry_safe is False + assert len(sent) == 1 + + +def test_fenced_msgpack_write_is_resent(cluster: Cluster) -> None: + client, sent = _fenced_client(cluster, fence_first_only=True, max_retries=2) + with client: + client.write.write_columnar( + "fence_mp", {"time": [int(time.time() * 1e6)], "host": ["a"], "v": [1.0]} + ) + assert len(sent) == 2 + + +# -------------------------------------------------------------------------- +# Primary writer hand-over (runs late: it moves the primary and restores it) +# -------------------------------------------------------------------------- + + +def test_demote_writer_refusals(cluster: Cluster) -> None: + primary = cluster.primary_id() + standby = cluster.standby_writer() + with pytest.raises(ArcServerError) as exc: + cluster.leader().cluster.demote_writer(standby, confirm=True) + assert exc.value.status_code == 409 + assert primary in str(exc.value) # Arc names the real primary + + with pytest.raises(ArcServerError) as exc: + cluster.follower().cluster.demote_writer(primary, confirm=True) + assert exc.value.status_code == 409 + assert "not the leader" in str(exc.value) + + +def _demote_and_wait(cluster: Cluster, node_id: str) -> str: + result = cluster.leader().cluster.demote_writer(node_id, confirm=True) + assert result["success"] is True + assert result["node_id"] == node_id + new_primary = result["new_primary"] + assert new_primary and new_primary != node_id + wait_for(cluster.primary_id, lambda p: p == new_primary, what="new primary in status") + cluster.wait_settled() + wait_for( + lambda: cluster.clients[new_primary].write_readiness().ready, + bool, + what="new primary ready for writes", + ) + return new_primary + + +def test_demote_primary_and_restore(cluster: Cluster) -> None: + cluster.wait_settled() + original = cluster.primary_id() + writers = cluster.writers() + if len(writers) < 2: + pytest.skip("needs two writers") + + new_primary = _demote_and_wait(cluster, original) + assert new_primary in writers + + # How the cluster reports it. + nodes = {n["id"]: n for n in cluster.any().cluster.nodes()["nodes"]} + assert nodes[new_primary]["writer_state"] == "primary" + assert nodes[original]["writer_state"] != "primary" + old = cluster.clients[original].write_readiness() + assert old.ready is False + assert old.status == "not_write_target" + # Writes still land through the old primary (forwarded to the new one). + db = unique("sdkcl_dm") + cluster.clients[original].write.write_line_protocol(["dm,host=a v=1"], database=db) + flush(cluster.clients[new_primary]) + wait_for( + lambda: cluster.leader().cluster.files(database=db)["files"], + bool, + what="write after hand-over in the manifest", + ) + assert {f["origin_node_id"] for f in cluster.leader().cluster.files(database=db)["files"]} == { + new_primary + } + + # Restore: there is no promote; demote until the original holds it again. + current = new_primary + for _ in range(len(writers) * 2): + if current == original: + break + current = _demote_and_wait(cluster, current) + assert cluster.primary_id() == original + + +# -------------------------------------------------------------------------- +# remove_node (opt-in; may require a cluster rebuild) +# -------------------------------------------------------------------------- + + +@needs_docker +@pytest.mark.skipif( + os.environ.get("ARC_TEST_CLUSTER_REMOVE_NODE") != "1", + reason="Set ARC_TEST_CLUSTER_REMOVE_NODE=1 to remove (and try to rejoin) the reader.", +) +def test_remove_node_reader(cluster: Cluster) -> None: + readers = cluster.readers() + if not readers: + pytest.skip("no reader to remove") + target = readers[0] + leader = cluster.leader() + + with pytest.raises(ArcServerError) as exc: + cluster.clients[target].cluster.remove_node(target) + assert exc.value.status_code == 400 # self-removal is refused + + result = leader.cluster.remove_node(target) + assert result["success"] is True + assert result["node_id"] == target + wait_for( + lambda: {n["id"] for n in leader.cluster.nodes()["nodes"]}, + lambda ids: target not in ids, + what="node gone from the registry", + ) + + docker("restart", target) + wait_for( + lambda: {n["id"] for n in leader.cluster.nodes()["nodes"]}, + lambda ids: target in ids, + timeout=120, + what=f"{target} to rejoin after restart", + ) + cluster.wait_settled(timeout=120) From 8094de0476e5bbe709d271cba1fd0b2768b2f637 Mon Sep 17 00:00:00 2001 From: Ignacio Van Droogenbroeck <64545348+xe-nvdk@users.noreply.github.com> Date: Mon, 2 Nov 2026 10:56:40 -0600 Subject: [PATCH 04/10] fix: retry behaviour and spoke registration found by live testing Found running the SDK against an Arc main build (1943fe87): - query_measurement retried Arc's governance 429 as an ordinary GET (5 requests, ~3s) before raising. It now retries 503 only, like the query POSTs. - register_spoke(spoke_id) always failed: the default name "" is refused by Arc ("spoke name is required"). name now defaults to spoke_id. - reconciliation.get_run and the spoke status/ledger reads retried a permanent 503 (no reconciliation / no edge-sync transport on this node) with backoff before raising. They now raise at once. --- src/arc_client/http/base.py | 9 ++++++ src/arc_client/management/async_edge_sync.py | 16 ++++++++--- .../management/async_reconciliation.py | 5 +++- src/arc_client/management/edge_sync.py | 16 ++++++++--- src/arc_client/management/reconciliation.py | 5 +++- src/arc_client/query/async_executor.py | 9 +++++- src/arc_client/query/executor.py | 9 +++++- tests/unit/http/test_edge_sync_contracts.py | 28 +++++++++++++++++++ tests/unit/http/test_query_parity.py | 21 ++++++++++++++ .../http/test_reconciliation_contracts.py | 16 +++++++++++ 10 files changed, 122 insertions(+), 12 deletions(-) diff --git a/src/arc_client/http/base.py b/src/arc_client/http/base.py index 24c8ad1..69f1884 100644 --- a/src/arc_client/http/base.py +++ b/src/arc_client/http/base.py @@ -282,6 +282,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 diff --git a/src/arc_client/management/async_edge_sync.py b/src/arc_client/management/async_edge_sync.py index 4a46e5e..ec9dd3c 100644 --- a/src/arc_client/management/async_edge_sync.py +++ b/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: @@ -169,7 +172,7 @@ async def spoke_status(self) -> dict[str, Any]: ``skipped``, ``pending_bytes`` and, once anything has synced, ``last_synced_at`` / ``seconds_since_last_sync``. """ - 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 +184,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). diff --git a/src/arc_client/management/async_reconciliation.py b/src/arc_client/management/async_reconciliation.py index c06317d..374d0a0 100644 --- a/src/arc_client/management/async_reconciliation.py +++ b/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) diff --git a/src/arc_client/management/edge_sync.py b/src/arc_client/management/edge_sync.py index 379debd..2574dad 100644 --- a/src/arc_client/management/edge_sync.py +++ b/src/arc_client/management/edge_sync.py @@ -34,6 +34,7 @@ from arc_client.config import ClientConfig from arc_client.exceptions import ArcError, ArcValidationError +from arc_client.http.base import never_retry_status from arc_client.http.sync_http import SyncHTTPClient from arc_client.models.edge_sync import Spoke, SpokeSecret @@ -142,14 +143,16 @@ def get_spoke(self, spoke_id: str) -> Spoke: """Get one registered spoke (hub). 404 if not registered.""" return Spoke.model_validate(self._dict("get", f"{SPOKES}/{_segment('spoke_id', spoke_id)}")) - def register_spoke(self, spoke_id: str, name: str = "") -> SpokeSecret: + 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(self._dict("post", f"{SPOKES}/", json=body)) def rotate_spoke_secret(self, spoke_id: str, confirm: bool = False) -> SpokeSecret: @@ -225,7 +228,7 @@ def spoke_status(self) -> dict[str, Any]: ``skipped``, ``pending_bytes`` and, once anything has synced, ``last_synced_at`` / ``seconds_since_last_sync``. """ - return self._dict("get", f"{SPOKE_SYNC}/status") + return self._dict("get", f"{SPOKE_SYNC}/status", retry_if=never_retry_status) def spoke_ledger( self, limit: Optional[int] = None, state: Optional[str] = None @@ -237,7 +240,12 @@ def spoke_ledger( ``exported``, ``failed`` or ``skipped``. ``limit`` is 1-1000 (server default 100). """ - return self._dict("get", f"{SPOKE_SYNC}/ledger", params=_ledger_params(limit, state)) + return self._dict( + "get", + f"{SPOKE_SYNC}/ledger", + params=_ledger_params(limit, state), + retry_if=never_retry_status, + ) def run_sync(self) -> dict[str, Any]: """Run one network sync pass now and wait for its result (spoke). diff --git a/src/arc_client/management/reconciliation.py b/src/arc_client/management/reconciliation.py index 3e8c8fd..b0f785a 100644 --- a/src/arc_client/management/reconciliation.py +++ b/src/arc_client/management/reconciliation.py @@ -18,6 +18,7 @@ from arc_client.config import ClientConfig from arc_client.exceptions import ArcError, ArcValidationError +from arc_client.http.base import never_retry_status from arc_client.http.sync_http import SyncHTTPClient from arc_client.models.reconciliation import ReconciliationRun @@ -103,4 +104,6 @@ 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(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(self._json("get", path, retry_if=never_retry_status), path) diff --git a/src/arc_client/query/async_executor.py b/src/arc_client/query/async_executor.py index 6ea9417..d87e5a4 100644 --- a/src/arc_client/query/async_executor.py +++ b/src/arc_client/query/async_executor.py @@ -15,6 +15,7 @@ ArcValidationError, ) from arc_client.http.async_http import AsyncHTTPClient +from arc_client.http.base import retry_unavailable_only from arc_client.models.query import EstimateResponse, MeasurementInfo, QueryResponse from arc_client.query.arrow_response import QUERY_ID_HEADER, decode_arrow_response from arc_client.query.msgpack_response import decode_msgpack_query_response @@ -440,7 +441,13 @@ async def query_measurement( measurement, database or self._config.database, limit, offset, order_by, where ) try: - response = await self._http.get(f"/api/v1/query/{measurement}", params=params) + response = await self._http.get( + f"/api/v1/query/{measurement}", + params=params, + # Governed like the query POSTs: a 429 is a rate limit or + # quota (query.go checkQueryGovernance), not worth retrying. + retry_if=retry_unavailable_only, + ) return query_response_from_json(response) except (ArcServerError, ArcConnectionError) as e: raise ArcQueryError(f"Query failed: {e}", status_code=_status(e)) from e diff --git a/src/arc_client/query/executor.py b/src/arc_client/query/executor.py index 2b8a992..30f08bb 100644 --- a/src/arc_client/query/executor.py +++ b/src/arc_client/query/executor.py @@ -14,6 +14,7 @@ ArcServerError, ArcValidationError, ) +from arc_client.http.base import retry_unavailable_only from arc_client.http.sync_http import SyncHTTPClient from arc_client.models.query import EstimateResponse, MeasurementInfo, QueryResponse from arc_client.query.arrow_response import QUERY_ID_HEADER, decode_arrow_response @@ -472,7 +473,13 @@ def query_measurement( measurement, database or self._config.database, limit, offset, order_by, where ) try: - response = self._http.get(f"/api/v1/query/{measurement}", params=params) + response = self._http.get( + f"/api/v1/query/{measurement}", + params=params, + # Governed like the query POSTs: a 429 is a rate limit or + # quota (query.go checkQueryGovernance), not worth retrying. + retry_if=retry_unavailable_only, + ) return query_response_from_json(response) except (ArcServerError, ArcConnectionError) as e: raise ArcQueryError(f"Query failed: {e}", status_code=_status(e)) from e diff --git a/tests/unit/http/test_edge_sync_contracts.py b/tests/unit/http/test_edge_sync_contracts.py index 06446ff..cb00b2c 100644 --- a/tests/unit/http/test_edge_sync_contracts.py +++ b/tests/unit/http/test_edge_sync_contracts.py @@ -435,3 +435,31 @@ async def test_export_disabled_503(httpx_mock: HTTPXMock, async_mode: bool) -> N with pytest.raises(ArcServerError) as info: await call(client, "apply_ack", "/mnt/usb/b1") assert info.value.status_code == 503 + + +async def test_register_spoke_name_defaults_to_id(httpx_mock: HTTPXMock, async_mode: bool) -> None: + # Found live: Arc refuses an empty name (registry.go "spoke name is required"). + httpx_mock.add_response( + method="POST", url=f"{SPOKES}/", status_code=201, json={"spoke_id": "r1", "secret": "s"} + ) + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + await call(client, "register_spoke", "r1") + assert body_of(httpx_mock) == {"spoke_id": "r1", "name": "r1"} + + +@pytest.mark.parametrize("method", ["spoke_status", "spoke_ledger"]) +async def test_spoke_reads_do_not_retry_disabled_transport( + httpx_mock: HTTPXMock, async_mode: bool, method: str +) -> None: + # Found live: 503 "no edge sync transport is enabled" is configuration. + httpx_mock.add_response(status_code=503, json={"error": "no edge sync transport is enabled"}) + async with AsyncExitStack() as stack: + client = ( + await stack.enter_async_context(AsyncArcClient(token="t", max_retries=3)) + if async_mode + else stack.enter_context(ArcClient(token="t", max_retries=3)) + ) + with pytest.raises(ArcServerError): + await call(client, method) + assert len(httpx_mock.get_requests()) == 1 diff --git a/tests/unit/http/test_query_parity.py b/tests/unit/http/test_query_parity.py index ca19e9e..e7fcee8 100644 --- a/tests/unit/http/test_query_parity.py +++ b/tests/unit/http/test_query_parity.py @@ -410,3 +410,24 @@ async def test_compressed_query_responses_decode(async_mode: bool, wire: str) -> server.server_close() assert result.data == [["a", 1]] assert "gzip" in _CompressingHandler.seen[-1] + + +@MODES +async def test_query_measurement_governance_429_not_retried( + httpx_mock: HTTPXMock, async_mode: bool +) -> None: + # Found live: as a GET it used to retry Arc's governance 429 (5 requests). + httpx_mock.add_response(status_code=429, json={"error": "rate limit exceeded"}) + async with make_client(async_mode, max_retries=3) as client: + with pytest.raises(ArcRateLimitError): + await call(client.query.query_measurement("cpu")) + assert len(httpx_mock.get_requests()) == 1 + + +@MODES +async def test_query_measurement_retries_503(httpx_mock: HTTPXMock, async_mode: bool) -> None: + httpx_mock.add_response(status_code=503, headers={"Retry-After": "0"}, json={}) + httpx_mock.add_response(json=JSON_OK) + async with make_client(async_mode, max_retries=2) as client: + await call(client.query.query_measurement("cpu")) + assert len(httpx_mock.get_requests()) == 2 diff --git a/tests/unit/http/test_reconciliation_contracts.py b/tests/unit/http/test_reconciliation_contracts.py index cfefa42..0a989b5 100644 --- a/tests/unit/http/test_reconciliation_contracts.py +++ b/tests/unit/http/test_reconciliation_contracts.py @@ -177,3 +177,19 @@ async def test_get_run_requires_id(httpx_mock: HTTPXMock, async_mode: bool) -> N client = await enter_client(stack, async_mode) with pytest.raises(ArcValidationError): await call(client, "get_run", "") + + +async def test_get_run_does_not_retry_permanent_503( + httpx_mock: HTTPXMock, async_mode: bool +) -> None: + # Found live: standalone Arc answers 503 "not running" with no Retry-After. + httpx_mock.add_response(status_code=503, json={"error": "reconciliation is not running"}) + async with AsyncExitStack() as stack: + client = ( + await stack.enter_async_context(AsyncArcClient(token="t", max_retries=3)) + if async_mode + else stack.enter_context(ArcClient(token="t", max_retries=3)) + ) + with pytest.raises(ArcServerError): + await call(client, "get_run", 7) + assert len(httpx_mock.get_requests()) == 1 From f33576581f702d13e2f1ec5558795b78f0199ba4 Mon Sep 17 00:00:00 2001 From: Ignacio Van Droogenbroeck <64545348+xe-nvdk@users.noreply.github.com> Date: Mon, 2 Nov 2026 10:56:54 -0600 Subject: [PATCH 05/10] docs: document Arc behaviour observed in live testing Arrow dictionary-encoding thresholds, the CQ execute end_time limit and watermark_status on standalone, spoke ledger discovery, and what an RBAC-only token (permissions=[]) can and cannot do. --- CHANGELOG.md | 4 +++- src/arc_client/management/async_continuous_query.py | 7 ++++++- src/arc_client/management/async_edge_sync.py | 4 +++- src/arc_client/management/continuous_query.py | 7 ++++++- src/arc_client/management/edge_sync.py | 4 +++- src/arc_client/query/async_executor.py | 5 ++++- src/arc_client/query/executor.py | 5 ++++- 7 files changed, 29 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 339d046..c021562 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/src/arc_client/management/async_continuous_query.py b/src/arc_client/management/async_continuous_query.py index e07e5fd..1f52c63 100644 --- a/src/arc_client/management/async_continuous_query.py +++ b/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"] = ( diff --git a/src/arc_client/management/async_edge_sync.py b/src/arc_client/management/async_edge_sync.py index ec9dd3c..f68369c 100644 --- a/src/arc_client/management/async_edge_sync.py +++ b/src/arc_client/management/async_edge_sync.py @@ -170,7 +170,9 @@ 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", retry_if=never_retry_status) diff --git a/src/arc_client/management/continuous_query.py b/src/arc_client/management/continuous_query.py index 601ae6e..67f1657 100644 --- a/src/arc_client/management/continuous_query.py +++ b/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"] = ( diff --git a/src/arc_client/management/edge_sync.py b/src/arc_client/management/edge_sync.py index 2574dad..1833acc 100644 --- a/src/arc_client/management/edge_sync.py +++ b/src/arc_client/management/edge_sync.py @@ -226,7 +226,9 @@ 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 self._dict("get", f"{SPOKE_SYNC}/status", retry_if=never_retry_status) diff --git a/src/arc_client/query/async_executor.py b/src/arc_client/query/async_executor.py index d87e5a4..0fc93f8 100644 --- a/src/arc_client/query/async_executor.py +++ b/src/arc_client/query/async_executor.py @@ -266,7 +266,10 @@ async def query_arrow( dictionary: Ask Arc to dictionary-encode low-cardinality string columns (``x-arc-arrow-dictionary: true``); those columns arrive as ``dictionary``. Smaller on the wire for - repetitive tags. + repetitive tags. Arc decides per column: only results of at + least 256 rows, and columns with at most rows/4 (and 4096) + distinct values, are encoded; otherwise columns stay plain + ``utf8`` (internal/api/query_arrow_dict.go). compression: Arrow IPC buffer compression, ``"zstd"`` or ``"lz4"`` (``x-arc-arrow-compression``), decoded transparently by pyarrow. ``None`` (default) sends uncompressed buffers. diff --git a/src/arc_client/query/executor.py b/src/arc_client/query/executor.py index 30f08bb..55b83f8 100644 --- a/src/arc_client/query/executor.py +++ b/src/arc_client/query/executor.py @@ -277,7 +277,10 @@ def query_arrow( dictionary: Ask Arc to dictionary-encode low-cardinality string columns (``x-arc-arrow-dictionary: true``); those columns arrive as ``dictionary``. Smaller on the wire for - repetitive tags. + repetitive tags. Arc decides per column: only results of at + least 256 rows, and columns with at most rows/4 (and 4096) + distinct values, are encoded; otherwise columns stay plain + ``utf8`` (internal/api/query_arrow_dict.go). compression: Arrow IPC buffer compression, ``"zstd"`` or ``"lz4"`` (``x-arc-arrow-compression``), decoded transparently by pyarrow. ``None`` (default) sends uncompressed buffers. From d96b724d1365930489f641053cd398be7a81ae16 Mon Sep 17 00:00:00 2001 From: Ignacio Van Droogenbroeck <64545348+xe-nvdk@users.noreply.github.com> Date: Mon, 2 Nov 2026 10:57:44 -0600 Subject: [PATCH 06/10] fix(query): name the right package in the pandas/polars install hint --- src/arc_client/query/async_executor.py | 4 ++-- src/arc_client/query/executor.py | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/src/arc_client/query/async_executor.py b/src/arc_client/query/async_executor.py index 0fc93f8..8a3b137 100644 --- a/src/arc_client/query/async_executor.py +++ b/src/arc_client/query/async_executor.py @@ -197,7 +197,7 @@ async def query_pandas( except ImportError as e: raise ImportError( "pandas is required for query_pandas(). " - "Install it with: pip install arc-client[pandas]" + "Install it with: pip install arc-tsdb-client[pandas]" ) from e table = await self.query_arrow( @@ -238,7 +238,7 @@ async def query_polars( except ImportError as e: raise ImportError( "polars is required for query_polars(). " - "Install it with: pip install arc-client[polars]" + "Install it with: pip install arc-tsdb-client[polars]" ) from e table = await self.query_arrow( diff --git a/src/arc_client/query/executor.py b/src/arc_client/query/executor.py index 55b83f8..4ca38f4 100644 --- a/src/arc_client/query/executor.py +++ b/src/arc_client/query/executor.py @@ -208,7 +208,7 @@ def query_pandas( except ImportError as e: raise ImportError( "pandas is required for query_pandas(). " - "Install it with: pip install arc-client[pandas]" + "Install it with: pip install arc-tsdb-client[pandas]" ) from e table = self.query_arrow(sql, database, dictionary=dictionary, compression=compression) @@ -251,7 +251,7 @@ def query_polars( except ImportError as e: raise ImportError( "polars is required for query_polars(). " - "Install it with: pip install arc-client[polars]" + "Install it with: pip install arc-tsdb-client[polars]" ) from e table = self.query_arrow(sql, database, dictionary=dictionary, compression=compression) From 5137616a8fcfaa5cf59896c1211d22775f3cc8a1 Mon Sep 17 00:00:00 2001 From: Ignacio Van Droogenbroeck <64545348+xe-nvdk@users.noreply.github.com> Date: Mon, 2 Nov 2026 10:58:25 -0600 Subject: [PATCH 07/10] test(integration): mark the two Arc bugs found live as strict xfail (arc#1424, arc#1425) --- tests/integration/test_live_parity.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/tests/integration/test_live_parity.py b/tests/integration/test_live_parity.py index 8adc287..20fc6e1 100644 --- a/tests/integration/test_live_parity.py +++ b/tests/integration/test_live_parity.py @@ -752,6 +752,9 @@ def test_partial_write_buffered_reported_once(admin: ArcClient, db: str) -> None assert stored == 0, "a dropped batch was stored" +@pytest.mark.xfail( + strict=True, reason="Arc bug Basekick-Labs/arc#1424: columnar type conflict returns 500" +) def test_columnar_type_conflict_status(admin: ArcClient, db: str) -> None: """A mixed-type column in one columnar batch is a client error. @@ -836,6 +839,10 @@ def test_databases_delete_allow_unreachable(admin: ArcClient) -> None: admin.databases.get(name) +@pytest.mark.xfail( + strict=True, + reason="Arc bug Basekick-Labs/arc#1425: deleted database still listed on local storage", +) def test_deleted_database_not_listed(admin: ArcClient) -> None: """After ``databases.delete`` the database must not be listed any more. From 1f8b4f2377d91dd37d0547042d8de48606a429c9 Mon Sep 17 00:00:00 2001 From: Ignacio Van Droogenbroeck <64545348+xe-nvdk@users.noreply.github.com> Date: Mon, 2 Nov 2026 10:59:05 -0600 Subject: [PATCH 08/10] test(integration): wait for the default-mode restore itself --- tests/integration/test_live_cluster.py | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/tests/integration/test_live_cluster.py b/tests/integration/test_live_cluster.py index d6d01d8..f75c8f0 100644 --- a/tests/integration/test_live_cluster.py +++ b/tests/integration/test_live_cluster.py @@ -675,9 +675,17 @@ def latest() -> dict[str, Any]: ) assert final["status"] == "completed", final # Arc's default for a scoped backup on a cluster node is replace. + previous = latest()["started_at"] default = node.backup.restore(backup_id, confirm=True) assert default["mode"] == "replace" - wait_for(latest, lambda s: s.get("status") == "completed", what="default restore") + final = wait_for( + latest, + lambda s: s.get("started_at") != previous + and s.get("status") in ("completed", "failed"), + what="default restore", + ) + assert final["mode"] == "replace" + assert final["status"] == "completed", final # The restored data is in the manifest again. assert cluster.leader().cluster.files(database=db)["total"] >= 1 finally: From 07abdd5e26df173b89ac766340c45dd7e79e36ac Mon Sep 17 00:00:00 2001 From: Ignacio Van Droogenbroeck <64545348+xe-nvdk@users.noreply.github.com> Date: Mon, 2 Nov 2026 11:03:48 -0600 Subject: [PATCH 09/10] fix: deletes outlast Arc's waits; timeouts and refusals reported honestly Found on a live 4-node cluster (Arc main 1943fe87): - A non-dry-run delete can wait 30 s inside Arc (deleteLocalWait, and the database delete's peer flush), equal to the default client timeout, so a delete that ran was reported as a connection failure and its unreachable_files were lost. Row and database deletes now use at least 90 s; post/delete take a per-request timeout. - A read timeout or a connection lost mid-response no longer says 'Failed to connect': the request was sent and may have been applied. - cluster.remove_node raised nothing when Arc answered 200 with success=false (no clustering); it now raises ArcError and escapes the id. - Docs: data deletes must go to the primary writer (others refuse with 503, unlike writes); assign_compactor's 409 when no compactor lease is managed. --- src/arc_client/http/async_http.py | 7 +- src/arc_client/http/base.py | 14 ++- src/arc_client/http/sync_http.py | 7 +- src/arc_client/management/async_cluster.py | 24 ++++- src/arc_client/management/async_databases.py | 7 +- src/arc_client/management/async_delete.py | 7 +- src/arc_client/management/cluster.py | 23 ++++- src/arc_client/management/databases.py | 7 +- src/arc_client/management/delete.py | 11 +- tests/unit/http/test_delete_timeouts.py | 102 +++++++++++++++++++ 10 files changed, 193 insertions(+), 16 deletions(-) create mode 100644 tests/unit/http/test_delete_timeouts.py diff --git a/src/arc_client/http/async_http.py b/src/arc_client/http/async_http.py index df70882..8cacb1a 100644 --- a/src/arc_client/http/async_http.py +++ b/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 diff --git a/src/arc_client/http/base.py b/src/arc_client/http/base.py index 69f1884..899a8f8 100644 --- a/src/arc_client/http/base.py +++ b/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. diff --git a/src/arc_client/http/sync_http.py b/src/arc_client/http/sync_http.py index ea780bb..3a1d035 100644 --- a/src/arc_client/http/sync_http.py +++ b/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 diff --git a/src/arc_client/management/async_cluster.py b/src/arc_client/management/async_cluster.py index ab104bf..c92a072 100644 --- a/src/arc_client/management/async_cluster.py +++ b/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: diff --git a/src/arc_client/management/async_databases.py b/src/arc_client/management/async_databases.py index 69f6fff..e9154f8 100644 --- a/src/arc_client/management/async_databases.py +++ b/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: diff --git a/src/arc_client/management/async_delete.py b/src/arc_client/management/async_delete.py index c1dc5de..161395a 100644 --- a/src/arc_client/management/async_delete.py +++ b/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: diff --git a/src/arc_client/management/cluster.py b/src/arc_client/management/cluster.py index 6806681..701fe46 100644 --- a/src/arc_client/management/cluster.py +++ b/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: diff --git a/src/arc_client/management/databases.py b/src/arc_client/management/databases.py index c582ec8..c26d8b8 100644 --- a/src/arc_client/management/databases.py +++ b/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: diff --git a/src/arc_client/management/delete.py b/src/arc_client/management/delete.py index 8f22bcc..d1cc710 100644 --- a/src/arc_client/management/delete.py +++ b/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: diff --git a/tests/unit/http/test_delete_timeouts.py b/tests/unit/http/test_delete_timeouts.py new file mode 100644 index 0000000..03c70ea --- /dev/null +++ b/tests/unit/http/test_delete_timeouts.py @@ -0,0 +1,102 @@ +"""Deletes outlast Arc's 30 s internal waits; timeouts are reported honestly. + +Found live on a 4-node cluster: a non-dry-run delete with allow_unreachable +took 30.1 s inside Arc (internal/api/delete_records.go deleteLocalWait), the +same as the default client timeout, so a delete that ran was reported as +"Failed to connect". +""" + +from __future__ import annotations + +from typing import Any + +import httpx +import pytest +from pytest_httpx import HTTPXMock + +from arc_client.exceptions import ArcConnectionError +from arc_client.management.delete import DELETE_TIMEOUT_FLOOR + +from ._parity import MODES, call, make_client + +DELETE_OK = {"success": True, "deleted_count": 1, "affected_files": 1} + + +def read_timeout(request: httpx.Request) -> Any: + return request.extensions["timeout"]["read"] + + +@MODES +async def test_delete_gets_timeout_floor(httpx_mock: HTTPXMock, async_mode: bool) -> None: + httpx_mock.add_response(json=DELETE_OK) + async with make_client(async_mode, timeout=30.0, max_retries=0) as client: + await call( + client.delete.delete( + "db", "cpu", "host = 'a'", dry_run=False, confirm=True, allow_unreachable=True + ) + ) + assert read_timeout(httpx_mock.get_request()) == DELETE_TIMEOUT_FLOOR + + +@MODES +async def test_dry_run_keeps_client_timeout(httpx_mock: HTTPXMock, async_mode: bool) -> None: + httpx_mock.add_response(json=DELETE_OK) + async with make_client(async_mode, timeout=30.0, max_retries=0) as client: + await call(client.delete.delete("db", "cpu", "host = 'a'")) + assert read_timeout(httpx_mock.get_request()) == 30.0 + + +@MODES +async def test_larger_client_timeout_wins(httpx_mock: HTTPXMock, async_mode: bool) -> None: + httpx_mock.add_response(json=DELETE_OK) + async with make_client(async_mode, timeout=300.0, max_retries=0) as client: + await call(client.delete.delete("db", "cpu", "host = 'a'", dry_run=False, confirm=True)) + assert read_timeout(httpx_mock.get_request()) == 300.0 + + +@MODES +async def test_database_delete_gets_timeout_floor(httpx_mock: HTTPXMock, async_mode: bool) -> None: + httpx_mock.add_response(json={"message": "deleted", "files_deleted": 1}) + async with make_client(async_mode, timeout=30.0, max_retries=0) as client: + await call(client.databases.delete("olddb", confirm=True)) + assert read_timeout(httpx_mock.get_request()) == DELETE_TIMEOUT_FLOOR + + +@MODES +async def test_read_timeout_is_not_reported_as_connect_failure( + httpx_mock: HTTPXMock, async_mode: bool +) -> None: + httpx_mock.add_exception(httpx.ReadTimeout("timed out")) + async with make_client(async_mode, max_retries=0) as client: + with pytest.raises(ArcConnectionError) as info: + await call(client.delete.delete("db", "cpu", "host = 'a'", dry_run=False, confirm=True)) + message = str(info.value) + assert "Failed to connect" not in message + assert "may have been applied" in message + + +@MODES +async def test_connect_error_still_says_failed_to_connect( + httpx_mock: HTTPXMock, async_mode: bool +) -> None: + httpx_mock.add_exception(httpx.ConnectError("refused")) + async with make_client(async_mode, max_retries=0) as client: + with pytest.raises(ArcConnectionError, match="Failed to connect"): + await call(client.health()) + + +@MODES +async def test_remove_node_refused_on_standalone_raises( + httpx_mock: HTTPXMock, async_mode: bool +) -> None: + # Found live: a node without clustering answers 200 {"success": false}, + # which the SDK used to return as if the node had been removed. + from arc_client.exceptions import ArcError + + httpx_mock.add_response( + json={"enabled": False, "success": False, "reason": "Clustering not enabled"} + ) + async with make_client(async_mode, max_retries=0) as client: + with pytest.raises(ArcError, match="not removed"): + await call(client.cluster.remove_node("node-a/b")) + assert httpx_mock.get_request().url.raw_path.endswith(b"/api/v1/cluster/nodes/node-a%2Fb") From ce884d8246b7f0df4db87204b42ed8287e6ce917 Mon Sep 17 00:00:00 2001 From: Ignacio Van Droogenbroeck <64545348+xe-nvdk@users.noreply.github.com> Date: Mon, 2 Nov 2026 11:16:35 -0600 Subject: [PATCH 10/10] test(integration): make the unreachable-file fixture deterministic Remove the file from every node, not just the primary: Arc's delete waits up to deleteLocalWait for listed files and the file puller heals a copy missing on one node from any peer, so the old fixture raced that pull. --- tests/integration/test_live_cluster.py | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) diff --git a/tests/integration/test_live_cluster.py b/tests/integration/test_live_cluster.py index f75c8f0..32e6623 100644 --- a/tests/integration/test_live_cluster.py +++ b/tests/integration/test_live_cluster.py @@ -540,7 +540,12 @@ def test_delete_on_standby_writer_is_refused_not_forwarded(cluster: Cluster) -> @pytest.fixture def missing_on_primary(cluster: Cluster) -> Iterator[tuple[str, str, str]]: - """A manifest file removed from the primary's disk (kept on the other nodes). + """A manifest file removed from every node's disk (kept in the manifest). + + It must be gone everywhere: Arc's delete waits up to deleteLocalWait for + listed files, and the file puller heals a copy missing on one node from + any peer that still has it, so a file removed from the primary alone is + only unreachable until that pull lands (the earlier, flaky version). Yields ``(database, measurement, path)``. Teardown removes the path from the manifest cluster-wide with delete_file. @@ -549,9 +554,10 @@ def missing_on_primary(cluster: Cluster) -> Iterator[tuple[str, str, str]]: pytest.skip("needs the docker CLI") db = unique("sdkcl_unr") path = write_and_locate(cluster, db, "unr", ["unr,host=a v=1", "unr,host=b v=2"])[0]["path"] - primary = cluster.primary_id() - docker("exec", primary, "rm", "-f", f"{DATA_DIR}/{path}") - assert not exists_on(primary, path) + for node in cluster.clients: + docker("exec", node, "rm", "-f", f"{DATA_DIR}/{path}") + for node in cluster.clients: + assert not exists_on(node, path) yield db, "unr", path with contextlib.suppress(ArcNotFoundError): cluster.leader().cluster.delete_file(path, confirm=True, reason="sdk test cleanup")