data: trace analysus multi ingestor support (#721)

* data: trace analysus multi ingestor support

* address review comments

.github/workflows/README.md

@@ -13,12 +13,3 @@
 - **`mobile.yml`** - Flutter mobile tests with coverage reporting
 - **`release.yml`** - Tag-triggered Flutter release builds for Android and iOS
 
-## Usage
-
-```bash
-# Build locally
-docker-compose build
-
-# Deploy
-docker-compose up -d
-```

data/mesh_ingestor/__init__.py

@@ -70,6 +70,7 @@ _CONFIG_ATTRS = {
     "CHANNEL_INDEX",
     "DEBUG",
     "INSTANCE",
+    "INSTANCES",
     "API_TOKEN",
     "ALLOWED_CHANNELS",
     "HIDDEN_CHANNELS",

data/mesh_ingestor/config.py

@@ -129,6 +129,11 @@ def _resolve_instance_domain() -> str:
 
     Reads the :envvar:`INSTANCE_DOMAIN` variable. When the value does not
     contain a scheme, ``https://`` is prepended automatically.
+
+    .. note::
+
+        Kept for backward compatibility with existing tests and callers.
+        New code should use :func:`_resolve_instance_domains` instead.
     """
 
     configured_instance = os.environ.get("INSTANCE_DOMAIN", "").rstrip("/")
@@ -139,8 +144,80 @@ def _resolve_instance_domain() -> str:
     return configured_instance
 
 
-INSTANCE = _resolve_instance_domain()
-API_TOKEN = os.environ.get("API_TOKEN", "")
+def _normalise_domain(raw: str) -> str:
+    """Strip whitespace and trailing slashes, prepend ``https://`` when needed.
+
+    Parameters:
+        raw: Single domain string to normalise.
+
+    Returns:
+        A URL string with a scheme prefix.
+    """
+
+    domain = raw.strip().rstrip("/")
+    if domain and "://" not in domain:
+        return f"https://{domain}"
+    return domain
+
+
+def _resolve_instance_domains() -> tuple[tuple[str, str], ...]:
+    """Parse :envvar:`INSTANCE_DOMAIN` and :envvar:`API_TOKEN` into paired tuples.
+
+    When ``INSTANCE_DOMAIN`` contains comma-separated values, each entry is
+    treated as an independent target.  ``API_TOKEN`` is either broadcast to
+    every target (single value) or positionally paired (comma-separated with
+    a matching count).
+
+    Returns:
+        A tuple of ``(instance_url, api_token)`` pairs, deduplicated by URL.
+
+    Raises:
+        ValueError: When the number of comma-separated tokens exceeds the
+            number of domains.
+    """
+
+    raw_domain = os.environ.get("INSTANCE_DOMAIN", "")
+    raw_token = os.environ.get("API_TOKEN", "")
+
+    domains: list[str] = []
+    seen: set[str] = set()
+    for part in raw_domain.split(","):
+        normalised = _normalise_domain(part)
+        if not normalised:
+            continue
+        key = normalised.casefold()
+        if key in seen:
+            continue
+        seen.add(key)
+        domains.append(normalised)
+
+    if not domains:
+        return ()
+
+    tokens = [t.strip() for t in raw_token.split(",")]
+    # A single token (including empty string) is broadcast to all domains.
+    if len(tokens) == 1:
+        token = tokens[0]
+        return tuple((d, token) for d in domains)
+
+    if len(tokens) != len(domains):
+        raise ValueError(
+            f"API_TOKEN has {len(tokens)} comma-separated values but "
+            f"INSTANCE_DOMAIN has {len(domains)}; counts must match or "
+            f"API_TOKEN must be a single value"
+        )
+
+    return tuple(zip(domains, tokens))
+
+
+INSTANCES: tuple[tuple[str, str], ...] = _resolve_instance_domains()
+"""Paired ``(instance_url, api_token)`` tuples derived from the environment."""
+
+INSTANCE = INSTANCES[0][0] if INSTANCES else _resolve_instance_domain()
+"""First configured instance URL, kept for backward compatibility."""
+
+API_TOKEN = INSTANCES[0][1] if INSTANCES else os.environ.get("API_TOKEN", "")
+"""API token for the first configured instance, kept for backward compatibility."""
 ENERGY_SAVING = os.environ.get("ENERGY_SAVING") == "1"
 """When ``True``, enables the ingestor's energy saving mode."""
 
@@ -202,6 +279,7 @@ __all__ = [
     "HIDDEN_CHANNELS",
     "ALLOWED_CHANNELS",
     "INSTANCE",
+    "INSTANCES",
     "API_TOKEN",
     "ENERGY_SAVING",
     "LORA_FREQ",

data/mesh_ingestor/daemon.py

@@ -666,11 +666,16 @@ def main(*, provider: MeshProtocol | None = None) -> None:
         signal.signal(signal.SIGINT, handle_sigint)
         signal.signal(signal.SIGTERM, handle_sigterm)
 
+    instance_label = (
+        ", ".join(inst for inst, _ in config.INSTANCES)
+        if config.INSTANCES
+        else "(no INSTANCE_DOMAIN configured)"
+    )
     config._debug_log(
         "Mesh daemon starting",
         context="daemon.main",
         severity="info",
-        target=config.INSTANCE or "(no INSTANCE_DOMAIN configured)",
+        target=instance_label,
         port=config.CONNECTION or "auto",
         channel=config.CHANNEL_INDEX,
     )

data/mesh_ingestor/queue.py

@@ -97,29 +97,24 @@ class QueueState:
 STATE = QueueState()
 
 
-def _post_json(
+def _send_single(
+    instance: str,
+    api_token: str,
     path: str,
     payload: dict,
-    *,
-    instance: str | None = None,
-    api_token: str | None = None,
 ) -> None:
-    """Send a JSON payload to the configured web API.
+    """Transmit a single JSON payload to one instance.
 
     Parameters:
-        path: API path relative to the configured instance root.
+        instance: Base URL of the target instance.
+        api_token: Bearer token for this instance (may be empty).
+        path: API path relative to the instance root.
         payload: JSON-serialisable body to transmit.
-        instance: Optional override for :data:`config.INSTANCE`.
-        api_token: Optional override for :data:`config.API_TOKEN`.
     """
 
-    if instance is None:
-        instance = config.INSTANCE
-    if api_token is None:
-        api_token = config.API_TOKEN
-
     if not instance:
         return
+
     url = f"{instance}{path}"
     data = json.dumps(payload).encode("utf-8")
 
@@ -155,6 +150,49 @@ def _post_json(
         )
 
 
+def _post_json(
+    path: str,
+    payload: dict,
+    *,
+    instance: str | None = None,
+    api_token: str | None = None,
+) -> None:
+    """Send a JSON payload to one or more configured web API instances.
+
+    When ``instance`` is provided explicitly the payload is sent to that
+    single target.  Otherwise every ``(url, token)`` pair in
+    :data:`config.INSTANCES` receives the payload independently so that
+    one failure does not block delivery to the remaining targets.
+
+    Parameters:
+        path: API path relative to the instance root.
+        payload: JSON-serialisable body to transmit.
+        instance: Optional single-instance override.
+        api_token: Optional token override (only used with ``instance``).
+    """
+
+    if instance is not None:
+        if not instance:
+            return
+        _send_single(instance, api_token or "", path, payload)
+        return
+
+    targets: tuple[tuple[str, str], ...] = config.INSTANCES
+    if not targets:
+        # Backward-compatible fallback for callers that only set
+        # config.INSTANCE / config.API_TOKEN directly.
+        inst = config.INSTANCE
+        if not inst:
+            return
+        _send_single(inst, api_token or config.API_TOKEN, path, payload)
+        return
+
+    for inst, token in targets:
+        if not inst:
+            continue
+        _send_single(inst, token, path, payload)
+
+
 def _enqueue_post_json(
     path: str,
     payload: dict,

tests/test_config_unit.py

@@ -96,6 +96,84 @@ class TestParseHiddenChannels:
 # ---------------------------------------------------------------------------
 
 
+class TestResolveInstanceDomains:
+    """Tests for :func:`config._resolve_instance_domains`."""
+
+    def test_single_domain(self, monkeypatch):
+        """Single domain produces one-element tuple."""
+        monkeypatch.setenv("INSTANCE_DOMAIN", "foo.tld")
+        monkeypatch.setenv("API_TOKEN", "secret")
+        result = config._resolve_instance_domains()
+        assert result == (("https://foo.tld", "secret"),)
+
+    def test_multi_domain_broadcast_token(self, monkeypatch):
+        """Multiple domains with a single token broadcast the token."""
+        monkeypatch.setenv("INSTANCE_DOMAIN", "foo.tld, bar.tld")
+        monkeypatch.setenv("API_TOKEN", "shared")
+        result = config._resolve_instance_domains()
+        assert result == (
+            ("https://foo.tld", "shared"),
+            ("https://bar.tld", "shared"),
+        )
+
+    def test_multi_domain_per_instance_tokens(self, monkeypatch):
+        """Comma-separated tokens are positionally paired with domains."""
+        monkeypatch.setenv("INSTANCE_DOMAIN", "a.tld,b.tld")
+        monkeypatch.setenv("API_TOKEN", "tok1,tok2")
+        result = config._resolve_instance_domains()
+        assert result == (("https://a.tld", "tok1"), ("https://b.tld", "tok2"))
+
+    def test_token_count_mismatch_raises(self, monkeypatch):
+        """Mismatched counts raise ValueError at parse time."""
+        monkeypatch.setenv("INSTANCE_DOMAIN", "a.tld,b.tld")
+        monkeypatch.setenv("API_TOKEN", "t1,t2,t3")
+        with pytest.raises(ValueError, match="counts must match"):
+            config._resolve_instance_domains()
+
+    def test_deduplicates_domains(self, monkeypatch):
+        """Duplicate domains are collapsed to a single entry."""
+        monkeypatch.setenv("INSTANCE_DOMAIN", "foo.tld, foo.tld")
+        monkeypatch.setenv("API_TOKEN", "tok")
+        result = config._resolve_instance_domains()
+        assert result == (("https://foo.tld", "tok"),)
+
+    def test_preserves_explicit_scheme(self, monkeypatch):
+        """Domains with explicit schemes keep them; others get https://."""
+        monkeypatch.setenv("INSTANCE_DOMAIN", "http://local:41447,bar.tld")
+        monkeypatch.setenv("API_TOKEN", "tok")
+        result = config._resolve_instance_domains()
+        assert result == (
+            ("http://local:41447", "tok"),
+            ("https://bar.tld", "tok"),
+        )
+
+    def test_empty_domain(self, monkeypatch):
+        """Empty INSTANCE_DOMAIN returns an empty tuple."""
+        monkeypatch.setenv("INSTANCE_DOMAIN", "")
+        monkeypatch.setenv("API_TOKEN", "tok")
+        result = config._resolve_instance_domains()
+        assert result == ()
+
+    def test_strips_trailing_slashes(self, monkeypatch):
+        """Trailing slashes are stripped from domains."""
+        monkeypatch.setenv("INSTANCE_DOMAIN", "foo.tld/")
+        monkeypatch.setenv("API_TOKEN", "tok")
+        result = config._resolve_instance_domains()
+        assert result == (("https://foo.tld", "tok"),)
+
+    def test_empty_token_broadcast(self, monkeypatch):
+        """Empty API_TOKEN broadcasts empty string to all instances."""
+        monkeypatch.setenv("INSTANCE_DOMAIN", "a.tld,b.tld")
+        monkeypatch.setenv("API_TOKEN", "")
+        result = config._resolve_instance_domains()
+        assert result == (("https://a.tld", ""), ("https://b.tld", ""))
+
+
+# ---------------------------------------------------------------------------
+# _resolve_instance_domain (legacy, kept for backward compatibility)
+# ---------------------------------------------------------------------------
+
+
 class TestResolveInstanceDomain:
     """Tests for :func:`config._resolve_instance_domain`."""
 

tests/test_mesh.py

@@ -233,7 +233,9 @@ def test_instance_domain_prefers_primary_env(mesh_module, monkeypatch):
     monkeypatch.setenv("INSTANCE_DOMAIN", "https://new.example")
 
     try:
+        refreshed_instances = mesh_module.config._resolve_instance_domains()
         refreshed_instance = mesh_module.config._resolve_instance_domain()
+        mesh_module.config.INSTANCES = refreshed_instances
         mesh_module.config.INSTANCE = refreshed_instance
         mesh_module.INSTANCE = refreshed_instance
 
@@ -241,6 +243,7 @@ def test_instance_domain_prefers_primary_env(mesh_module, monkeypatch):
         assert mesh_module.INSTANCE == "https://new.example"
     finally:
         monkeypatch.delenv("INSTANCE_DOMAIN", raising=False)
+        mesh_module.config.INSTANCES = mesh_module.config._resolve_instance_domains()
         mesh_module.config.INSTANCE = mesh_module.config._resolve_instance_domain()
         mesh_module.INSTANCE = mesh_module.config.INSTANCE
 
@@ -251,7 +254,9 @@ def test_instance_domain_infers_scheme_for_hostnames(mesh_module, monkeypatch):
     monkeypatch.setenv("INSTANCE_DOMAIN", "mesh.example.org")
 
     try:
+        refreshed_instances = mesh_module.config._resolve_instance_domains()
         refreshed_instance = mesh_module.config._resolve_instance_domain()
+        mesh_module.config.INSTANCES = refreshed_instances
         mesh_module.config.INSTANCE = refreshed_instance
         mesh_module.INSTANCE = refreshed_instance
 
@@ -259,6 +264,7 @@ def test_instance_domain_infers_scheme_for_hostnames(mesh_module, monkeypatch):
         assert mesh_module.INSTANCE == "https://mesh.example.org"
     finally:
         monkeypatch.delenv("INSTANCE_DOMAIN", raising=False)
+        mesh_module.config.INSTANCES = mesh_module.config._resolve_instance_domains()
         mesh_module.config.INSTANCE = mesh_module.config._resolve_instance_domain()
         mesh_module.INSTANCE = mesh_module.config.INSTANCE
 

tests/test_queue_unit.py

@@ -53,6 +53,19 @@ def _fresh_state() -> QueueState:
     return QueueState()
 
 
+class _FakeResp:
+    """Minimal context-manager response stub for ``urlopen`` patches."""
+
+    def read(self):
+        return b""
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *a):
+        pass
+
+
 # ---------------------------------------------------------------------------
 # Priority constant ordering
 # ---------------------------------------------------------------------------
@@ -85,33 +98,24 @@ class TestPostJson:
     """Tests for :func:`queue._post_json`."""
 
     def test_skips_when_no_instance(self, monkeypatch):
-        """Does nothing when INSTANCE is empty."""
+        """Does nothing when INSTANCES is empty."""
+        monkeypatch.setattr(config, "INSTANCES", ())
         monkeypatch.setattr(config, "INSTANCE", "")
-        sent = []
         with patch("urllib.request.urlopen") as mock_open:
             _post_json("/api/test", {"key": "val"})
             mock_open.assert_not_called()
 
     def test_sends_json_post(self, monkeypatch):
         """Sends a POST request with JSON body and correct headers."""
+        monkeypatch.setattr(config, "INSTANCES", (("http://localhost", "tok"),))
         monkeypatch.setattr(config, "INSTANCE", "http://localhost")
         monkeypatch.setattr(config, "API_TOKEN", "tok")
 
         captured_req = []
 
-        class FakeResp:
-            def read(self):
-                return b""
-
-            def __enter__(self):
-                return self
-
-            def __exit__(self, *a):
-                pass
-
         def fake_urlopen(req, timeout=None):
             captured_req.append(req)
-            return FakeResp()
+            return _FakeResp()
 
         with patch("urllib.request.urlopen", fake_urlopen):
             _post_json("/api/nodes", {"a": 1})
@@ -124,6 +128,7 @@ class TestPostJson:
 
     def test_handles_network_error_gracefully(self, monkeypatch, capsys):
         """Network errors are caught and logged, not raised."""
+        monkeypatch.setattr(config, "INSTANCES", (("http://localhost", ""),))
         monkeypatch.setattr(config, "INSTANCE", "http://localhost")
         monkeypatch.setattr(config, "API_TOKEN", "")
         monkeypatch.setattr(config, "DEBUG", True)
@@ -140,19 +145,9 @@ class TestPostJson:
 
         captured_req = []
 
-        class FakeResp:
-            def read(self):
-                return b""
-
-            def __enter__(self):
-                return self
-
-            def __exit__(self, *a):
-                pass
-
         def fake_urlopen(req, timeout=None):
             captured_req.append(req)
-            return FakeResp()
+            return _FakeResp()
 
         with patch("urllib.request.urlopen", fake_urlopen):
             _post_json("/api/test", {}, instance="http://override")
@@ -161,24 +156,15 @@ class TestPostJson:
 
     def test_no_auth_header_when_token_empty(self, monkeypatch):
         """No Authorization header is added when API_TOKEN is empty."""
+        monkeypatch.setattr(config, "INSTANCES", (("http://localhost", ""),))
         monkeypatch.setattr(config, "INSTANCE", "http://localhost")
         monkeypatch.setattr(config, "API_TOKEN", "")
 
         captured_req = []
 
-        class FakeResp:
-            def read(self):
-                return b""
-
-            def __enter__(self):
-                return self
-
-            def __exit__(self, *a):
-                pass
-
         def fake_urlopen(req, timeout=None):
             captured_req.append(req)
-            return FakeResp()
+            return _FakeResp()
 
         with patch("urllib.request.urlopen", fake_urlopen):
             _post_json("/api/test", {})
@@ -394,3 +380,106 @@ class TestClearPostQueue:
         state = _fresh_state()
         _clear_post_queue(state=state)
         assert state.queue == []
+
+
+# ---------------------------------------------------------------------------
+# Multi-instance fan-out
+# ---------------------------------------------------------------------------
+
+
+class TestMultiInstanceFanOut:
+    """Tests for multi-instance POST fan-out in :func:`queue._post_json`."""
+
+    def test_fans_out_to_all_instances(self, monkeypatch):
+        """Each configured instance receives the payload."""
+        monkeypatch.setattr(
+            config,
+            "INSTANCES",
+            (("http://alpha", "t1"), ("http://beta", "t2")),
+        )
+
+        captured = []
+
+        def fake_urlopen(req, timeout=None):
+            captured.append(req)
+            return _FakeResp()
+
+        with patch("urllib.request.urlopen", fake_urlopen):
+            _post_json("/api/nodes", {"a": 1})
+
+        assert len(captured) == 2
+        urls = {r.get_full_url() for r in captured}
+        assert urls == {"http://alpha/api/nodes", "http://beta/api/nodes"}
+        tokens = {r.get_header("Authorization") for r in captured}
+        assert tokens == {"Bearer t1", "Bearer t2"}
+
+    def test_failure_isolation(self, monkeypatch):
+        """A failure on one instance does not prevent delivery to the next."""
+        monkeypatch.setattr(
+            config,
+            "INSTANCES",
+            (("http://broken", "t1"), ("http://ok", "t2")),
+        )
+        monkeypatch.setattr(config, "DEBUG", False)
+
+        captured = []
+
+        def fake_urlopen(req, timeout=None):
+            if "broken" in req.get_full_url():
+                raise OSError("connection refused")
+            captured.append(req)
+            return _FakeResp()
+
+        with patch("urllib.request.urlopen", fake_urlopen):
+            _post_json("/api/test", {"x": 1})
+
+        assert len(captured) == 1
+        assert "http://ok" in captured[0].get_full_url()
+
+    def test_explicit_instance_skips_fanout(self, monkeypatch):
+        """Passing instance= explicitly bypasses the INSTANCES fan-out."""
+        monkeypatch.setattr(
+            config,
+            "INSTANCES",
+            (("http://a", "t1"), ("http://b", "t2")),
+        )
+
+        captured = []
+
+        def fake_urlopen(req, timeout=None):
+            captured.append(req)
+            return _FakeResp()
+
+        with patch("urllib.request.urlopen", fake_urlopen):
+            _post_json("/api/test", {}, instance="http://override")
+
+        assert len(captured) == 1
+        assert "http://override" in captured[0].get_full_url()
+
+    def test_empty_instances_noop(self, monkeypatch):
+        """No requests are made when INSTANCES is empty."""
+        monkeypatch.setattr(config, "INSTANCES", ())
+        monkeypatch.setattr(config, "INSTANCE", "")
+
+        with patch("urllib.request.urlopen") as mock_open:
+            _post_json("/api/test", {})
+            mock_open.assert_not_called()
+
+    def test_backward_compat_fallback(self, monkeypatch):
+        """Falls back to config.INSTANCE when INSTANCES is empty."""
+        monkeypatch.setattr(config, "INSTANCES", ())
+        monkeypatch.setattr(config, "INSTANCE", "http://legacy")
+        monkeypatch.setattr(config, "API_TOKEN", "tok")
+
+        captured = []
+
+        def fake_urlopen(req, timeout=None):
+            captured.append(req)
+            return _FakeResp()
+
+        with patch("urllib.request.urlopen", fake_urlopen):
+            _post_json("/api/test", {"v": 1})
+
+        assert len(captured) == 1
+        assert "http://legacy" in captured[0].get_full_url()
+        assert captured[0].get_header("Authorization") == "Bearer tok"