test(integration): shared-storage live suite + stream claims from a killed primary

CHANGELOG.md

@@ -119,7 +119,10 @@ plus bug fixes and the first real HTTP-layer test coverage.
 - Real HTTP round-trip test coverage (via `pytest-httpx`) for headers, error
   mapping, compression on the wire, retries, msgpack decode, imports, databases,
   and enterprise wrappers — sync and async. Opt-in live integration tests under
-  `tests/integration` (gated on `ARC_TEST_URL`).
+  `tests/integration`: a single node (`ARC_TEST_URL`), a local-storage cluster
+  (`ARC_TEST_CLUSTER_URLS`, with an opt-in stream-takeover claim test), and a
+  shared-storage cluster with a compactor node, a cold tier and MQTT
+  (`ARC_TEST_SHARED_URLS`).
 
 ### Changed
 

src/arc_client/management/async_delete.py

@@ -46,7 +46,10 @@ async def delete(
                 DELETE runs anyway and the rows in those files or on those
                 nodes are NOT deleted; the response names them in
                 ``unreachable_files`` and ``unflushed_nodes``. Sent only when
-                True. Needs Arc 2027.01.1+; older servers ignore it.
+                True. Needs Arc 2027.01.1+; older servers ignore it. Either
+                way Arc first waits for those files or nodes (about 30-40 s
+                for a node that does not answer); a real delete gets a
+                request timeout of at least 90 s to outlast that.
         """
         if not database:
             raise ArcValidationError("database is required")

src/arc_client/management/async_mqtt.py

@@ -49,6 +49,11 @@ async def get_subscription(self, subscription_id: str) -> Any:
     async def create_subscription(self, config: dict[str, Any]) -> Any:
         """Create a subscription.
 
+        A new subscription is ``stopped``: call :meth:`start_subscription`.
+        ``auto_start`` starts it when Arc boots, not on create. In a cluster a
+        subscription lives on the node it was created on (that node's
+        database), and only that node lists and runs it.
+
         Args:
             config: The subscription body (name, broker, client_id, topics, qos,
                 database, tls_*, auto_start, topic_mapping, etc.). See the Arc

src/arc_client/management/async_tiering.py

@@ -40,7 +40,12 @@ async def files(
         database: Optional[str] = None,
         limit: Optional[int] = None,
     ) -> Any:
-        """List tiered files, optionally filtered by tier/database."""
+        """List tiered files, optionally filtered by tier/database.
+
+        Arc 2027.01.1 ignores ``tier`` when ``database`` is also given and
+        returns the database's files in every tier (Basekick-Labs/arc#1430);
+        filter on each file's ``tier`` to be safe.
+        """
         params = {
             k: v
             for k, v in {"tier": tier, "database": database, "limit": limit}.items()

src/arc_client/management/delete.py

@@ -53,7 +53,10 @@ def delete(
                 DELETE runs anyway and the rows in those files or on those
                 nodes are NOT deleted; the response names them in
                 ``unreachable_files`` and ``unflushed_nodes``. Sent only when
-                True. Needs Arc 2027.01.1+; older servers ignore it.
+                True. Needs Arc 2027.01.1+; older servers ignore it. Either
+                way Arc first waits for those files or nodes (about 30-40 s
+                for a node that does not answer); a real delete gets a
+                request timeout of at least 90 s to outlast that.
         """
         if not database:
             raise ArcValidationError("database is required")

src/arc_client/management/mqtt.py

@@ -49,6 +49,11 @@ def get_subscription(self, subscription_id: str) -> Any:
     def create_subscription(self, config: dict[str, Any]) -> Any:
         """Create a subscription.
 
+        A new subscription is ``stopped``: call :meth:`start_subscription`.
+        ``auto_start`` starts it when Arc boots, not on create. In a cluster a
+        subscription lives on the node it was created on (that node's
+        database), and only that node lists and runs it.
+
         Args:
             config: The subscription body (name, broker, client_id, topics, qos,
                 database, tls_*, auto_start, topic_mapping, etc.). See the Arc

src/arc_client/management/tiering.py

@@ -40,7 +40,12 @@ def files(
         database: Optional[str] = None,
         limit: Optional[int] = None,
     ) -> Any:
-        """List tiered files, optionally filtered by tier/database."""
+        """List tiered files, optionally filtered by tier/database.
+
+        Arc 2027.01.1 ignores ``tier`` when ``database`` is also given and
+        returns the database's files in every tier (Basekick-Labs/arc#1430);
+        filter on each file's ``tier`` to be safe.
+        """
         params = {
             k: v
             for k, v in {"tier": tier, "database": database, "limit": limit}.items()

src/arc_client/models/delete.py

@@ -20,7 +20,8 @@ class DeleteResponse(BaseModel):
         files_processed: List of processed file names.
         failed_files: Files whose deletion could not be completed.
         unreachable_files: Manifest files unavailable on this node.
-        unflushed_nodes: Nodes whose buffered rows could not be flushed.
+        unflushed_nodes: Nodes whose buffered rows could not be flushed, one
+            message per node naming it and why (not a bare node id).
         cold_deleted_files: Whole cold files deleted or selected by a dry run.
         cold_files: Partially matching cold files left unchanged.
         cold_rows_not_deleted: Matching rows left in cold storage.

tests/integration/test_live_cluster.py

@@ -21,6 +21,10 @@
 ``docker`` CLI is not available. ``ARC_TEST_CLUSTER_DATA_DIR`` overrides the
 in-container storage root (default ``/app/data/arc``).
 
+``test_stream_claim_after_primary_killed`` kills the primary writer, so it runs
+only with ``ARC_TEST_CLUSTER_STREAM_TAKEOVER=1`` on a cluster with stream
+takeover enabled (see its docstring).
+
 ``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.
@@ -36,6 +40,7 @@
 import uuid
 from collections.abc import Iterator
 from dataclasses import dataclass, field
+from datetime import datetime, timezone
 from typing import Any, Callable, Optional, TypeVar
 from urllib.parse import urlparse
 
@@ -136,6 +141,13 @@ def exists_on(container: str, path: str) -> bool:
     return rc == 0
 
 
+def _docker_time(value: str) -> float:
+    """Seconds since the epoch of a ``docker inspect`` RFC 3339 timestamp."""
+    head, _, frac = value.rstrip("Z").partition(".")
+    stamp = datetime.strptime(head, "%Y-%m-%dT%H:%M:%S").replace(tzinfo=timezone.utc)
+    return stamp.timestamp() + float(f"0.{frac or 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")
@@ -807,8 +819,8 @@ def test_stream_claims_listing(cluster: Cluster) -> None:
 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
+    unfinished = [c for c in claims if not c["done"] and not c["released"]]
+    if unfinished:  # pragma: no cover - needs a claimant that cannot finish
         claim = unfinished[0]
         result = leader.cluster.release_stream_claim(claim["instance"], claim["epoch"])
         assert result["success"] is True
@@ -825,6 +837,89 @@ def test_release_stream_claim(cluster: Cluster) -> None:
     assert "leader" in str(exc.value)
 
 
+@needs_docker
+@pytest.mark.skipif(
+    os.environ.get("ARC_TEST_CLUSTER_STREAM_TAKEOVER") != "1",
+    reason="Set ARC_TEST_CLUSTER_STREAM_TAKEOVER=1 (stream-takeover cluster; kills the primary).",
+)
+def test_stream_claim_after_primary_killed(cluster: Cluster) -> None:
+    """A standby claims what a killed primary acknowledged but never flushed.
+
+    Needs ``cluster.wal_coverage`` and ``cluster.stream_takeover`` on every
+    node, automatic failover, and an ``ingest.max_buffer_age_ms`` long
+    enough (e.g. 120000) that the rows are still unflushed when the
+    primary dies. Kills the primary container and starts it again.
+    """
+    db = unique("sdkcl_claim")
+
+    def least_uptime() -> float:
+        started = [
+            docker("inspect", "-f", "{{.State.StartedAt}}", w).strip() for w in cluster.writers()
+        ]
+        now = time.time()
+        return min(now - _docker_time(s) for s in started)
+
+    # A writer restarted moments ago (an earlier run of this test) is not yet
+    # receiving the primary's stream, holds no copy, and may be the one
+    # promoted. Give every standby time to connect first.
+    wait_for(least_uptime, lambda up: up >= 45, timeout=90, interval=3, what="writers up 45s")
+    primary = cluster.primary_id()
+    listing = cluster.clients[primary].cluster.stream_claims()
+    assert listing["enabled"] is True
+    # Earlier kills leave finished claims of the same node id behind.
+    before = {(c["instance"], c["epoch"]) for c in listing["claims"]}
+    cluster.clients[primary].write.write_line_protocol(
+        [f"st,host=h{i} v={i}" for i in range(5)], database=db
+    )
+    time.sleep(1.5)  # the origin stream reaches the standbys in milliseconds
+    docker("kill", primary)
+    try:
+        survivors = {nid: c for nid, c in cluster.clients.items() if nid != primary}
+        survivor = next(iter(survivors.values()))
+
+        def claims() -> list[dict[str, Any]]:
+            listed = survivor.cluster.stream_claims()["claims"]
+            return [
+                c
+                for c in listed
+                if c["origin"] == primary and (c["instance"], c["epoch"]) not in before
+            ]
+
+        claim = wait_for(
+            claims,
+            lambda cs: bool(cs) and all(c["done"] for c in cs),
+            timeout=120,
+            interval=2,
+            what="a finished stream claim of the killed primary",
+        )[-1]
+        assert claim["by"] != primary
+        assert claim["batches"] >= 1
+        assert claim["unreplayable_batches"] == 0
+        # The claimed rows are published once, not lost and not duplicated.
+        for client in survivors.values():
+            rows = client.query.query("SELECT count(*) FROM st", database=db).data
+            assert rows[0][0] == 5
+
+        leader_id = survivor.cluster.status()["raft"]["leader_id"]
+        leader = cluster.clients[leader_id]
+        for kwargs in ({}, {"abandon": True, "confirm": True}):
+            with pytest.raises(ArcServerError) as exc:
+                leader.cluster.release_stream_claim(claim["instance"], claim["epoch"], **kwargs)
+            assert exc.value.status_code == 409
+            assert "finished" in str(exc.value)
+        with pytest.raises(ArcNotFoundError):
+            leader.cluster.release_stream_claim(claim["instance"], claim["epoch"] + 1)
+        follower = next(c for nid, c in survivors.items() if nid != leader_id)
+        with pytest.raises(ArcServerError) as exc:
+            follower.cluster.release_stream_claim(claim["instance"], claim["epoch"])
+        assert exc.value.status_code == 409
+    finally:
+        docker("start", primary)
+        cluster.wait_settled(timeout=180)
+        with contextlib.suppress(Exception):
+            cluster.clients[cluster.primary_id()].databases.delete(db, confirm=True)
+
+
 def test_abort_compaction_job_unknown(cluster: Cluster) -> None:
     status = cluster.status()
     wedged = status.get("compaction_commit", {}).get("wedged_jobs") or []