diff --git a/.github/workflows/README.md b/.github/workflows/README.md index 65534bf..e6b8661 100644 --- a/.github/workflows/README.md +++ b/.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 -``` diff --git a/data/mesh_ingestor/__init__.py b/data/mesh_ingestor/__init__.py index f4af0b7..dfea05e 100644 --- a/data/mesh_ingestor/__init__.py +++ b/data/mesh_ingestor/__init__.py @@ -70,6 +70,7 @@ _CONFIG_ATTRS = { "CHANNEL_INDEX", "DEBUG", "INSTANCE", + "INSTANCES", "API_TOKEN", "ALLOWED_CHANNELS", "HIDDEN_CHANNELS", diff --git a/data/mesh_ingestor/config.py b/data/mesh_ingestor/config.py index eefd70c..6429397 100644 --- a/data/mesh_ingestor/config.py +++ b/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", diff --git a/data/mesh_ingestor/daemon.py b/data/mesh_ingestor/daemon.py index 1bb7cbe..3c5a09b 100644 --- a/data/mesh_ingestor/daemon.py +++ b/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, ) diff --git a/data/mesh_ingestor/queue.py b/data/mesh_ingestor/queue.py index 3d1af4e..12ab3f0 100644 --- a/data/mesh_ingestor/queue.py +++ b/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, diff --git a/tests/test_config_unit.py b/tests/test_config_unit.py index 4ea9afc..b675862 100644 --- a/tests/test_config_unit.py +++ b/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`.""" diff --git a/tests/test_mesh.py b/tests/test_mesh.py index 964496b..a763984 100644 --- a/tests/test_mesh.py +++ b/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 diff --git a/tests/test_queue_unit.py b/tests/test_queue_unit.py index 71352ae..54d3ae2 100644 --- a/tests/test_queue_unit.py +++ b/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"