diff --git a/README.md b/README.md index 33caea7..2972aa4 100644 --- a/README.md +++ b/README.md @@ -191,10 +191,17 @@ with wildedge.track(handle): |---|---|---|---| | `dsn` | `-` | `WILDEDGE_DSN` | Required. `https://@ingest.wildedge.dev/` | | `app_version` | `None` | `-` | Optional. Your app's version string. | +| `app_identity` | `` | `WILDEDGE_APP_IDENTITY` | Namespace for offline persistence paths. Set per-app to isolate multi-process workloads in one project. | | `debug` | `false` | `WILDEDGE_DEBUG` | Log events to console. | | `batch_size` | `10` | `-` | Events per transmission (recommended: 1-100). | | `flush_interval_sec` | `60` | `-` | Max seconds between flushes (recommended: 1-3600). | | `max_queue_size` | `200` | `-` | In-memory buffer limit (recommended: 10-10000). | +| `enable_offline_persistence` | `true` | `-` | Persist pending unsent events on disk and replay on restart. | +| `offline_queue_dir` | OS-specific state dir | `-` | Folder for pending queue persistence (defaults to platform state path). | +| `max_event_age_sec` | `900` | `-` | Max age for queued events before dead-lettering. | +| `enable_dead_letter_persistence` | `false` | `-` | Persist dropped batches/events to disk dead-letter store. | +| `dead_letter_dir` | OS-specific cache dir | `-` | Directory where dead-letter batch files are stored. | +| `max_dead_letter_batches` | `10` | `-` | Max dead-letter batch files retained on disk. | ## Testing diff --git a/scripts/run_compat_local.py b/scripts/run_compat_local.py old mode 100644 new mode 100755 diff --git a/tests/test_batch.py b/tests/test_batch.py index f9f691d..efb5ff7 100644 --- a/tests/test_batch.py +++ b/tests/test_batch.py @@ -2,7 +2,7 @@ from datetime import datetime, timezone -from wildedge import config +from wildedge import constants from wildedge.batch import build_batch from wildedge.device import DeviceInfo @@ -24,7 +24,7 @@ def test_returns_protocol_version(self): session_id="sess-1", created_at=datetime.now(timezone.utc), ) - assert batch["protocol_version"] == config.PROTOCOL_VERSION + assert batch["protocol_version"] == constants.PROTOCOL_VERSION def test_includes_device(self): batch = build_batch( @@ -58,6 +58,24 @@ def test_includes_events(self): ) assert batch["events"] == events + def test_internal_queue_fields_are_not_sent(self): + events = [ + { + "event_type": "inference", + "__we_first_queued_at": 1.0, + "__we_attempts": 3, + } + ] + batch = build_batch( + device=make_device(), + models={}, + events=events, + session_id="sess-1", + created_at=datetime.now(timezone.utc), + ) + assert "__we_first_queued_at" not in batch["events"][0] + assert "__we_attempts" not in batch["events"][0] + def test_batch_id_is_unique(self): now = datetime.now(timezone.utc) b1 = build_batch(make_device(), {}, [], "s", now) diff --git a/tests/test_cli.py b/tests/test_cli.py index 9a173b3..9b76e9f 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -6,7 +6,7 @@ import pytest -from wildedge import cli +from wildedge import cli, constants from wildedge.integrations.registry import IntegrationSpec from wildedge.runtime import bootstrap from wildedge.runtime import runner as runtime_runner @@ -59,6 +59,9 @@ def fake_run(cmd, env, check): # type: ignore[no-untyped-def] assert captured["env"][bootstrap.RUN_PROPAGATE_ENV] == "1" assert captured["env"][bootstrap.RUN_STRICT_INTEGRATIONS_ENV] == "0" assert captured["env"][bootstrap.RUN_PRINT_STARTUP_REPORT_ENV] == "0" + assert captured["env"][bootstrap.RUN_FLUSH_TIMEOUT_ENV] == str( + constants.DEFAULT_SHUTDOWN_FLUSH_TIMEOUT_SEC + ) def test_cli_run_sets_no_propagate_and_strict(monkeypatch): @@ -126,6 +129,34 @@ def test_install_runtime_requires_dsn(monkeypatch): bootstrap.install_runtime() +def test_install_runtime_default_flush_timeout_is_shutdown_budget(monkeypatch): + class FakeWildEdge: + SUPPORTED_INTEGRATIONS = {"onnx"} + + def __init__(self, *, dsn, app_version, debug): # type: ignore[no-untyped-def] + pass + + def instrument(self, name): # type: ignore[no-untyped-def] + pass + + def flush(self, timeout): # type: ignore[no-untyped-def] + pass + + def close(self): # type: ignore[no-untyped-def] + pass + + monkeypatch.setattr(bootstrap, "WildEdge", FakeWildEdge) + monkeypatch.setenv(bootstrap.RUN_DSN_ENV, "https://secret@ingest.wildedge.dev/key") + monkeypatch.delenv(bootstrap.RUN_FLUSH_TIMEOUT_ENV, raising=False) + monkeypatch.setattr(bootstrap.importlib.util, "find_spec", lambda _: object()) + + context = bootstrap.install_runtime() + try: + assert context.flush_timeout == constants.DEFAULT_SHUTDOWN_FLUSH_TIMEOUT_SEC + finally: + context.shutdown() + + def test_install_runtime_instruments_requested_integrations(monkeypatch): events: list[tuple[str, str]] = [] @@ -248,6 +279,53 @@ def test_doctor_runtime_config_fail(monkeypatch, capsys): assert "runtime_config: FAIL (batch_size out of range)" in out +def test_doctor_reports_offline_and_dead_letter_checks(monkeypatch, capsys): + monkeypatch.setenv("WILDEDGE_DSN", "https://secret@ingest.wildedge.dev/key") + monkeypatch.setattr(cli.importlib.util, "find_spec", lambda _: object()) + monkeypatch.setattr(cli, "check_writable_dir", lambda _: (True, "ok")) + rc = cli.main(["doctor", "--integrations", "onnx"]) + out = capsys.readouterr().out + assert rc == 0 + assert "offline_queue_capacity: OK" in out + assert "dead_letter_capacity: OK" in out + assert "writable_offline_queue_dir: OK (ok)" in out + assert "writable_dead_letter_dir: SKIP" in out + + +def test_doctor_reports_dead_letter_dir_when_enabled(monkeypatch, capsys): + monkeypatch.setenv("WILDEDGE_DSN", "https://secret@ingest.wildedge.dev/key") + monkeypatch.setattr(cli.importlib.util, "find_spec", lambda _: object()) + monkeypatch.setattr(cli, "check_writable_dir", lambda _: (True, "ok")) + rc = cli.main(["doctor", "--integrations", "onnx", "--dead-letter-persistence"]) + out = capsys.readouterr().out + assert rc == 0 + assert "writable_dead_letter_dir: OK (ok)" in out + + +def test_doctor_uses_project_key_for_default_namespace(monkeypatch, capsys): + monkeypatch.setenv("WILDEDGE_DSN", "https://secret@ingest.wildedge.dev/test-prod") + monkeypatch.setattr(cli.importlib.util, "find_spec", lambda _: object()) + monkeypatch.setattr(cli, "check_writable_dir", lambda path: (True, str(path))) + monkeypatch.delenv("WILDEDGE_APP_IDENTITY", raising=False) + rc = cli.main(["doctor", "--integrations", "onnx"]) + out = capsys.readouterr().out + assert rc == 0 + assert "/test-prod/pending_queue" in out + assert "/test-prod/dead_letters" in out + + +def test_doctor_uses_app_identity_override_for_namespace(monkeypatch, capsys): + monkeypatch.setenv("WILDEDGE_DSN", "https://secret@ingest.wildedge.dev/test-prod") + monkeypatch.setenv("WILDEDGE_APP_IDENTITY", "my-app") + monkeypatch.setattr(cli.importlib.util, "find_spec", lambda _: object()) + monkeypatch.setattr(cli, "check_writable_dir", lambda path: (True, str(path))) + rc = cli.main(["doctor", "--integrations", "onnx"]) + out = capsys.readouterr().out + assert rc == 0 + assert "/my-app/pending_queue" in out + assert "/my-app/dead_letters" in out + + def test_runner_clears_runtime_env_when_no_propagate(monkeypatch): class FakeContext: debug = False diff --git a/tests/test_client.py b/tests/test_client.py index 23e0ba7..971c64a 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -4,7 +4,7 @@ import pytest -from wildedge import config +from wildedge import constants @pytest.fixture(autouse=True) @@ -15,6 +15,7 @@ def mock_dependencies(): patch("wildedge.client.Transmitter"), patch("wildedge.client.Consumer"), patch("wildedge.client.EventQueue"), + patch("wildedge.client.DeadLetterStore"), patch("wildedge.client.ModelRegistry"), ): yield @@ -25,7 +26,7 @@ def test_batch_size_too_low(): with pytest.raises( ValueError, - match=f"batch_size must be between {config.BATCH_SIZE_MIN} and {config.BATCH_SIZE_MAX}", + match=f"batch_size must be between {constants.BATCH_SIZE_MIN} and {constants.BATCH_SIZE_MAX}", ): WildEdge(dsn="https://test@test.com/key", batch_size=0) @@ -35,7 +36,7 @@ def test_batch_size_too_high(): with pytest.raises( ValueError, - match=f"batch_size must be between {config.BATCH_SIZE_MIN} and {config.BATCH_SIZE_MAX}", + match=f"batch_size must be between {constants.BATCH_SIZE_MIN} and {constants.BATCH_SIZE_MAX}", ): WildEdge(dsn="https://test@test.com/key", batch_size=101) @@ -45,7 +46,7 @@ def test_flush_interval_too_low(): with pytest.raises( ValueError, - match=f"flush_interval_sec must be between {config.FLUSH_INTERVAL_MIN} and {config.FLUSH_INTERVAL_MAX}", + match=f"flush_interval_sec must be between {constants.FLUSH_INTERVAL_MIN} and {constants.FLUSH_INTERVAL_MAX}", ): WildEdge(dsn="https://test@test.com/key", flush_interval_sec=0) @@ -55,7 +56,7 @@ def test_flush_interval_too_high(): with pytest.raises( ValueError, - match=f"flush_interval_sec must be between {config.FLUSH_INTERVAL_MIN} and {config.FLUSH_INTERVAL_MAX}", + match=f"flush_interval_sec must be between {constants.FLUSH_INTERVAL_MIN} and {constants.FLUSH_INTERVAL_MAX}", ): WildEdge(dsn="https://test@test.com/key", flush_interval_sec=3601) @@ -65,7 +66,7 @@ def test_max_queue_size_too_low(): with pytest.raises( ValueError, - match=f"max_queue_size must be between {config.MAX_QUEUE_SIZE_MIN} and {config.MAX_QUEUE_SIZE_MAX}", + match=f"max_queue_size must be between {constants.MAX_QUEUE_SIZE_MIN} and {constants.MAX_QUEUE_SIZE_MAX}", ): WildEdge(dsn="https://test@test.com/key", max_queue_size=9) @@ -75,7 +76,7 @@ def test_max_queue_size_too_high(): with pytest.raises( ValueError, - match=f"max_queue_size must be between {config.MAX_QUEUE_SIZE_MIN} and {config.MAX_QUEUE_SIZE_MAX}", + match=f"max_queue_size must be between {constants.MAX_QUEUE_SIZE_MIN} and {constants.MAX_QUEUE_SIZE_MAX}", ): WildEdge(dsn="https://test@test.com/key", max_queue_size=10001) @@ -91,3 +92,72 @@ def test_valid_values(): max_queue_size=500, ) assert client is not None + + +def test_max_event_age_must_be_positive(): + from wildedge.client import WildEdge + + with pytest.raises(ValueError, match="max_event_age_sec must be greater than 0"): + WildEdge(dsn="https://test@test.com/key", max_event_age_sec=0) + + +def test_max_dead_letter_batches_must_be_non_negative(): + from wildedge.client import WildEdge + + with pytest.raises(ValueError, match="max_dead_letter_batches must be >= 0"): + WildEdge(dsn="https://test@test.com/key", max_dead_letter_batches=-1) + + +def test_app_identity_defaults_to_project_key(): + from wildedge.client import WildEdge + + with ( + patch( + "wildedge.client.default_pending_queue_dir", return_value="pending-dir" + ) as p, + patch("wildedge.client.default_dead_letter_dir", return_value="dead-dir") as d, + patch( + "wildedge.client.default_model_registry_path", return_value="registry-path" + ) as r, + ): + WildEdge(dsn="https://test@test.com/proj-key") + p.assert_called_once_with("proj-key") + d.assert_called_once_with("proj-key") + r.assert_called_once_with("proj-key") + + +def test_app_identity_override_used_for_paths(): + from wildedge.client import WildEdge + + with ( + patch( + "wildedge.client.default_pending_queue_dir", return_value="pending-dir" + ) as p, + patch("wildedge.client.default_dead_letter_dir", return_value="dead-dir") as d, + patch( + "wildedge.client.default_model_registry_path", return_value="registry-path" + ) as r, + ): + WildEdge(dsn="https://test@test.com/proj-key", app_identity="app-a") + p.assert_called_once_with("app-a") + d.assert_called_once_with("app-a") + r.assert_called_once_with("app-a") + + +def test_app_identity_env_override_used_for_paths(monkeypatch): + from wildedge.client import WildEdge + + monkeypatch.setenv(constants.ENV_APP_IDENTITY, "env-app") + with ( + patch( + "wildedge.client.default_pending_queue_dir", return_value="pending-dir" + ) as p, + patch("wildedge.client.default_dead_letter_dir", return_value="dead-dir") as d, + patch( + "wildedge.client.default_model_registry_path", return_value="registry-path" + ) as r, + ): + WildEdge(dsn="https://test@test.com/proj-key") + p.assert_called_once_with("env-app") + d.assert_called_once_with("env-app") + r.assert_called_once_with("env-app") diff --git a/tests/test_consumer.py b/tests/test_consumer.py index 5ed4d21..29acc1b 100644 --- a/tests/test_consumer.py +++ b/tests/test_consumer.py @@ -2,6 +2,7 @@ from unittest.mock import MagicMock +from wildedge import constants from wildedge.consumer import Consumer from wildedge.device import DeviceInfo from wildedge.queue import EventQueue @@ -97,6 +98,77 @@ def test_400_discards_events(self): consumer.drain_once() assert queue.length() == 0 # Events discarded on 400 + def test_permanent_error_discards_events(self): + queue = EventQueue(max_size=100) + queue.add({"event_id": "e1", "event_type": "inference", "model_id": "m"}) + + mock_transmitter = MagicMock(spec=Transmitter) + mock_transmitter.send.return_value = IngestResponse( + status="error", + batch_id="b-1", + events_accepted=0, + events_rejected=1, + ) + + consumer = self._make_consumer(queue, mock_transmitter) + consumer.drain_once() + assert queue.length() == 0 # Permanent transmitter error is discarded + + def test_permanent_error_persists_dead_letter_and_calls_callback(self): + queue = EventQueue(max_size=100) + queue.add({"event_id": "e1", "event_type": "inference", "model_id": "m"}) + dead_letter_store = MagicMock() + on_failure = MagicMock() + + mock_transmitter = MagicMock(spec=Transmitter) + mock_transmitter.send.return_value = IngestResponse( + status="error", + batch_id="b-1", + events_accepted=0, + events_rejected=1, + ) + + consumer = self._make_consumer( + queue, + mock_transmitter, + dead_letter_store=dead_letter_store, + on_delivery_failure=on_failure, + ) + consumer.drain_once() + + assert queue.length() == 0 + assert dead_letter_store.write.called + reason = dead_letter_store.write.call_args.kwargs["reason"] + assert reason == "permanent_error" + assert on_failure.called + assert on_failure.call_args_list[0].args == ("permanent_error", 1, 0) + + def test_max_event_age_drops_stale_events_without_transmit(self): + queue = EventQueue(max_size=100) + queue.add( + { + "event_id": "e1", + "event_type": "inference", + "model_id": "m", + "__we_first_queued_at": 1.0, + } + ) + dead_letter_store = MagicMock() + + mock_transmitter = MagicMock(spec=Transmitter) + + consumer = self._make_consumer( + queue, + mock_transmitter, + max_event_age_sec=0.01, + dead_letter_store=dead_letter_store, + ) + consumer.drain_once() + + assert queue.length() == 0 + mock_transmitter.send.assert_not_called() + assert dead_letter_store.write.called + def test_batch_includes_model_registry(self): queue = EventQueue(max_size=100) queue.add({"event_id": "e1", "event_type": "inference", "model_id": "my-model"}) @@ -128,3 +200,138 @@ def test_drain_once_returns_false_when_empty(self): result = consumer.drain_once() assert result is False mock_transmitter.send.assert_not_called() + + def test_flush_does_not_tight_loop_on_transmit_error(self, monkeypatch): + monkeypatch.setattr(Consumer, "run", lambda self: None) + queue = EventQueue(max_size=100) + queue.add({"event_id": "e1", "event_type": "inference", "model_id": "m"}) + mock_transmitter = MagicMock(spec=Transmitter) + mock_transmitter.send.side_effect = TransmitError("Network error") + + consumer = self._make_consumer(queue, mock_transmitter) + consumer.flush(timeout=0.2) + assert mock_transmitter.send.call_count == 1 + + def test_next_retry_delay_scales_and_caps(self): + queue = EventQueue(max_size=100) + mock_transmitter = MagicMock(spec=Transmitter) + consumer = self._make_consumer(queue, mock_transmitter) + + delay, next_backoff = consumer.next_retry_delay(1.0, jitter=False) + assert delay == 1.0 + assert next_backoff == 2.0 + + delay, next_backoff = consumer.next_retry_delay(1.0, jitter=False, max_wait=0.3) + assert delay == 0.3 + assert next_backoff == 2.0 + + def test_next_retry_delay_applies_jitter(self, monkeypatch): + queue = EventQueue(max_size=100) + mock_transmitter = MagicMock(spec=Transmitter) + consumer = self._make_consumer(queue, mock_transmitter) + + monkeypatch.setattr("wildedge.consumer.random.uniform", lambda a, b: 0.2) + delay, _ = consumer.next_retry_delay(1.0, jitter=True) + assert delay == 1.2 + + def test_run_uses_shared_retry_delay(self, monkeypatch): + original_run = Consumer.run + monkeypatch.setattr(Consumer, "run", lambda self: None) + queue = EventQueue(max_size=100) + queue.add({"event_id": "e1", "event_type": "inference", "model_id": "m"}) + mock_transmitter = MagicMock(spec=Transmitter) + consumer = self._make_consumer(queue, mock_transmitter) + consumer.run = original_run.__get__(consumer, Consumer) + + class StopControl: + def __init__(self): + self.flag = False + self.wait_calls: list[float] = [] + + def is_set(self) -> bool: + return self.flag + + def wait(self, timeout=None) -> None: + self.wait_calls.append(timeout) + self.flag = True + + def set(self) -> None: + self.flag = True + + stop_control = StopControl() + consumer.stop_event = stop_control + consumer.drain_once = lambda: False + + called = {"count": 0} + + def fake_next_retry_delay(backoff, *, jitter, max_wait=None): + called["count"] += 1 + assert backoff == 1.0 + assert jitter is True + assert max_wait is None + return 0.42, 2.0 + + consumer.next_retry_delay = fake_next_retry_delay + consumer.run() + assert called["count"] == 1 + assert stop_control.wait_calls[0] == 0.42 + + def test_flush_uses_shared_retry_delay(self, monkeypatch): + monkeypatch.setattr(Consumer, "run", lambda self: None) + queue = EventQueue(max_size=100) + queue.add({"event_id": "e1", "event_type": "inference", "model_id": "m"}) + mock_transmitter = MagicMock(spec=Transmitter) + mock_transmitter.send.side_effect = TransmitError("Network error") + + consumer = self._make_consumer(queue, mock_transmitter) + calls = {"count": 0} + + def fake_next_retry_delay(backoff, *, jitter, max_wait=None): + calls["count"] += 1 + assert jitter is True + assert max_wait is not None + return max_wait, backoff + + consumer.next_retry_delay = fake_next_retry_delay + consumer.flush(timeout=0.15) + assert calls["count"] >= 1 + + def test_close_default_is_best_effort_non_blocking(self, monkeypatch): + monkeypatch.setattr(Consumer, "run", lambda self: None) + queue = EventQueue(max_size=100) + queue.add({"event_id": "e1", "event_type": "inference", "model_id": "m"}) + mock_transmitter = MagicMock(spec=Transmitter) + mock_transmitter.send.side_effect = TransmitError("Network error") + + consumer = self._make_consumer(queue, mock_transmitter) + consumer.close() + assert mock_transmitter.send.call_count == 0 + assert queue.length() == 1 + + def test_close_with_timeout_attempts_flush(self, monkeypatch): + monkeypatch.setattr(Consumer, "run", lambda self: None) + queue = EventQueue(max_size=100) + queue.add({"event_id": "e1", "event_type": "inference", "model_id": "m"}) + mock_transmitter = MagicMock(spec=Transmitter) + mock_transmitter.send.side_effect = TransmitError("Network error") + + consumer = self._make_consumer(queue, mock_transmitter) + consumer.close(timeout=0.15) + assert mock_transmitter.send.call_count >= 1 + + def test_atexit_registers_shutdown_flush_budget(self, monkeypatch): + registered = {} + + def fake_register(fn, *args, **kwargs): + registered["fn"] = fn + registered["args"] = args + registered["kwargs"] = kwargs + + monkeypatch.setattr("wildedge.consumer.atexit.register", fake_register) + monkeypatch.setattr(Consumer, "run", lambda self: None) + + queue = EventQueue(max_size=100) + mock_transmitter = MagicMock(spec=Transmitter) + consumer = self._make_consumer(queue, mock_transmitter) + assert registered["fn"] == consumer.flush + assert registered["args"] == (constants.DEFAULT_SHUTDOWN_FLUSH_TIMEOUT_SEC,) diff --git a/tests/test_dead_letters.py b/tests/test_dead_letters.py new file mode 100644 index 0000000..e779e7e --- /dev/null +++ b/tests/test_dead_letters.py @@ -0,0 +1,38 @@ +from __future__ import annotations + +from wildedge.dead_letters import DeadLetterStore + + +def test_dead_letter_store_writes_batch_file(tmp_path): + store = DeadLetterStore( + enabled=True, + directory=str(tmp_path), + max_batches=10, + ) + store.write(reason="test", events=[{"event_id": "e1"}], batch_id="b1") + files = list(tmp_path.glob("*.json")) + assert len(files) == 1 + + +def test_dead_letter_store_enforces_max_batches(tmp_path): + store = DeadLetterStore( + enabled=True, + directory=str(tmp_path), + max_batches=2, + ) + store.write(reason="test", events=[{"event_id": "e1"}], batch_id="b1") + store.write(reason="test", events=[{"event_id": "e2"}], batch_id="b2") + store.write(reason="test", events=[{"event_id": "e3"}], batch_id="b3") + files = sorted(tmp_path.glob("*.json")) + assert len(files) == 2 + + +def test_dead_letter_store_disabled_no_files(tmp_path): + store = DeadLetterStore( + enabled=False, + directory=str(tmp_path), + max_batches=10, + ) + store.write(reason="test", events=[{"event_id": "e1"}], batch_id="b1") + files = list(tmp_path.glob("*.json")) + assert len(files) == 0 diff --git a/tests/test_offline_replay.py b/tests/test_offline_replay.py new file mode 100644 index 0000000..c98c502 --- /dev/null +++ b/tests/test_offline_replay.py @@ -0,0 +1,66 @@ +from __future__ import annotations + +from unittest.mock import patch + +from wildedge.client import WildEdge +from wildedge.device import DeviceInfo + + +class _DummyConsumer: + def __init__(self, *args, **kwargs): + pass + + def flush(self, timeout: float = 5.0) -> None: + pass + + def close(self, timeout: float | None = None) -> None: + pass + + +class _Model: + pass + + +def test_offline_replay_restores_model_registry_for_pending_events(tmp_path): + queue_dir = tmp_path / "queue" + dead_dir = tmp_path / "dead" + with ( + patch( + "wildedge.client.detect_device", + return_value=DeviceInfo(device_id="d", device_type="linux"), + ), + patch("wildedge.client.Transmitter"), + patch("wildedge.client.Consumer", _DummyConsumer), + ): + client_a = WildEdge( + dsn="https://secret@ingest.wildedge.dev/proj", + app_identity="app-a", + offline_queue_dir=str(queue_dir), + dead_letter_dir=str(dead_dir), + enable_offline_persistence=True, + ) + client_a.register_model( + _Model(), + model_id="ResNet", + source="local", + family="resnet", + version="1.0", + quantization="fp32", + ) + client_a.publish( + {"event_id": "e1", "event_type": "inference", "model_id": "ResNet"} + ) + client_a.close() + + client_b = WildEdge( + dsn="https://secret@ingest.wildedge.dev/proj", + app_identity="app-a", + offline_queue_dir=str(queue_dir), + dead_letter_dir=str(dead_dir), + enable_offline_persistence=True, + ) + + assert client_b.queue.length() == 1 + models = client_b.registry.snapshot() + assert "ResNet" in models + assert models["ResNet"]["model_name"] == "_Model" diff --git a/tests/test_platform_adapters.py b/tests/test_platform_adapters.py index 981287d..5af6286 100644 --- a/tests/test_platform_adapters.py +++ b/tests/test_platform_adapters.py @@ -68,3 +68,14 @@ def cuInit(self, _flags): # noqa: N802 monkeypatch.setattr("ctypes.CDLL", lambda _name: FakeCudaLib()) assert cuda_device_count("libcuda.so.1") == 0 + + +def test_platform_adapters_expose_state_and_cache_paths(): + for adapter in ( + LinuxPlatform(), + MacOSPlatform(), + WindowsPlatform(), + UnknownPlatform(), + ): + assert adapter.state_base() + assert adapter.cache_base() diff --git a/tests/test_queue.py b/tests/test_queue.py index a9360c8..9970e41 100644 --- a/tests/test_queue.py +++ b/tests/test_queue.py @@ -106,3 +106,35 @@ def add_events(): assert not errors assert q.length() <= 1000 + + def test_persistent_queue_rehydrates_from_disk(self, tmp_path): + q1 = EventQueue( + max_size=10, + policy=QueuePolicy.OPPORTUNISTIC, + persist_to_disk=True, + disk_dir=str(tmp_path), + ) + q1.add(make_event(1)) + q1.add(make_event(2)) + + q2 = EventQueue( + max_size=10, + policy=QueuePolicy.OPPORTUNISTIC, + persist_to_disk=True, + disk_dir=str(tmp_path), + ) + assert q2.length() == 2 + assert q2.peek() == make_event(1) + + def test_persistent_queue_remove_deletes_files(self, tmp_path): + q = EventQueue( + max_size=10, + policy=QueuePolicy.OPPORTUNISTIC, + persist_to_disk=True, + disk_dir=str(tmp_path), + ) + q.add(make_event(1)) + q.add(make_event(2)) + assert len(list(tmp_path.glob("*.json"))) == 2 + q.remove_first() + assert len(list(tmp_path.glob("*.json"))) == 1 diff --git a/tests/test_settings.py b/tests/test_settings.py new file mode 100644 index 0000000..b02aefc --- /dev/null +++ b/tests/test_settings.py @@ -0,0 +1,55 @@ +from __future__ import annotations + +from wildedge import constants +from wildedge.settings import ( + RUN_DSN_ENV, + RUN_FLUSH_TIMEOUT_ENV, + RUN_INTEGRATIONS_ENV, + parse_bool, + read_client_env, + read_runtime_env, + resolve_app_identity, +) + + +def test_parse_bool_variants(): + assert parse_bool("1") is True + assert parse_bool("true") is True + assert parse_bool("yes") is True + assert parse_bool("on") is True + assert parse_bool("0") is False + assert parse_bool(None) is False + + +def test_read_client_env_prefers_explicit_values(): + env = { + constants.ENV_DSN: "https://a@ingest.wildedge.dev/proj", + constants.ENV_DEBUG: "1", + constants.ENV_APP_IDENTITY: "env-app", + } + s = read_client_env( + dsn="https://b@ingest.wildedge.dev/proj2", + debug=False, + app_identity="explicit-app", + environ=env, + ) + assert s.dsn == "https://b@ingest.wildedge.dev/proj2" + assert s.debug is False + assert s.app_identity == "explicit-app" + + +def test_read_runtime_env_uses_run_over_base_dsn(): + env = { + constants.ENV_DSN: "https://base@ingest.wildedge.dev/base", + RUN_DSN_ENV: "https://run@ingest.wildedge.dev/run", + RUN_FLUSH_TIMEOUT_ENV: "7.5", + RUN_INTEGRATIONS_ENV: "onnx,timm", + } + s = read_runtime_env(all_integrations=["onnx", "timm"], environ=env) + assert s.dsn == "https://run@ingest.wildedge.dev/run" + assert s.flush_timeout == 7.5 + assert s.integrations == ["onnx", "timm"] + + +def test_resolve_app_identity_fallbacks_to_project_key(): + assert resolve_app_identity(explicit=None, project_key="proj", environ={}) == "proj" diff --git a/wildedge/batch.py b/wildedge/batch.py index 092a3b5..9afa3df 100644 --- a/wildedge/batch.py +++ b/wildedge/batch.py @@ -3,10 +3,14 @@ import uuid from datetime import datetime, timezone -from wildedge import config +from wildedge import constants from wildedge.device import DeviceInfo +def _sanitize_event(event: dict) -> dict: + return {k: v for k, v in event.items() if not k.startswith("__we_")} + + def build_batch( device: DeviceInfo, models: dict[str, dict], @@ -16,12 +20,12 @@ def build_batch( ) -> dict: """Build a protocol-compliant batch envelope.""" return { - "protocol_version": config.PROTOCOL_VERSION, + "protocol_version": constants.PROTOCOL_VERSION, "device": device.to_dict(), "models": models, "session_id": session_id, "batch_id": str(uuid.uuid4()), "created_at": created_at.isoformat(), "sent_at": datetime.now(timezone.utc).isoformat(), - "events": events, + "events": [_sanitize_event(event) for event in events], } diff --git a/wildedge/cli.py b/wildedge/cli.py index f9c4e7d..743e275 100644 --- a/wildedge/cli.py +++ b/wildedge/cli.py @@ -14,10 +14,11 @@ from pathlib import Path from urllib.parse import urlparse -from wildedge import config +from wildedge import constants from wildedge.client import parse_dsn from wildedge.device import get_device_id_path from wildedge.integrations.registry import INTEGRATIONS_BY_NAME, supported_integrations +from wildedge.paths import default_dead_letter_dir, default_pending_queue_dir from wildedge.runtime.bootstrap import ( RUN_APP_VERSION_ENV, RUN_DEBUG_ENV, @@ -28,6 +29,7 @@ RUN_PROPAGATE_ENV, RUN_STRICT_INTEGRATIONS_ENV, ) +from wildedge.settings import read_client_env, resolve_app_identity def build_parser() -> argparse.ArgumentParser: @@ -51,7 +53,7 @@ def build_parser() -> argparse.ArgumentParser: run.add_argument( "--flush-timeout", type=float, - default=5.0, + default=constants.DEFAULT_SHUTDOWN_FLUSH_TIMEOUT_SEC, help="Flush timeout (seconds) for shutdown.", ) run.add_argument( @@ -112,21 +114,76 @@ def build_parser() -> argparse.ArgumentParser: doctor.add_argument( "--batch-size", type=int, - default=config.DEFAULT_BATCH_SIZE, + default=constants.DEFAULT_BATCH_SIZE, help="Validate intended batch size against SDK limits.", ) doctor.add_argument( "--flush-interval", type=float, - default=config.DEFAULT_FLUSH_INTERVAL_SEC, + default=constants.DEFAULT_FLUSH_INTERVAL_SEC, help="Validate intended flush interval against SDK limits.", ) doctor.add_argument( "--max-queue-size", type=int, - default=config.DEFAULT_MAX_QUEUE_SIZE, + default=constants.DEFAULT_MAX_QUEUE_SIZE, help="Validate intended queue size against SDK limits.", ) + doctor.add_argument( + "--max-event-age-sec", + type=float, + default=constants.DEFAULT_MAX_EVENT_AGE_SEC, + help="Validate queued event age cap (seconds).", + ) + doctor.add_argument( + "--max-dead-letter-batches", + type=int, + default=constants.DEFAULT_MAX_DEAD_LETTER_BATCHES, + help="Validate dead-letter retention cap (batch files).", + ) + doctor.add_argument( + "--offline-queue-dir", + default=None, + help="Offline queue persistence directory.", + ) + doctor.add_argument( + "--dead-letter-dir", + default=None, + help="Dead-letter directory.", + ) + doctor.add_argument( + "--app-identity", + default=None, + help="Namespace for offline/dead-letter paths (defaults to DSN project key).", + ) + offline_persistence = doctor.add_mutually_exclusive_group() + offline_persistence.add_argument( + "--offline-persistence", + dest="offline_persistence", + action="store_true", + default=constants.DEFAULT_ENABLE_OFFLINE_PERSISTENCE, + help="Enable offline queue persistence checks.", + ) + offline_persistence.add_argument( + "--no-offline-persistence", + dest="offline_persistence", + action="store_false", + help="Disable offline queue persistence checks.", + ) + dead_letter_persistence = doctor.add_mutually_exclusive_group() + dead_letter_persistence.add_argument( + "--dead-letter-persistence", + dest="dead_letter_persistence", + action="store_true", + default=constants.DEFAULT_ENABLE_DEAD_LETTER_PERSISTENCE, + help="Enable dead-letter persistence checks.", + ) + dead_letter_persistence.add_argument( + "--no-dead-letter-persistence", + dest="dead_letter_persistence", + action="store_false", + help="Disable dead-letter persistence checks.", + ) return parser @@ -219,16 +276,24 @@ def check_writable_dir(path: Path) -> tuple[bool, str]: def validate_runtime_config(parsed: argparse.Namespace) -> tuple[bool, str]: - if not (config.BATCH_SIZE_MIN <= parsed.batch_size <= config.BATCH_SIZE_MAX): + if not (constants.BATCH_SIZE_MIN <= parsed.batch_size <= constants.BATCH_SIZE_MAX): return False, "batch_size out of range" if not ( - config.FLUSH_INTERVAL_MIN <= parsed.flush_interval <= config.FLUSH_INTERVAL_MAX + constants.FLUSH_INTERVAL_MIN + <= parsed.flush_interval + <= constants.FLUSH_INTERVAL_MAX ): return False, "flush_interval out of range" if not ( - config.MAX_QUEUE_SIZE_MIN <= parsed.max_queue_size <= config.MAX_QUEUE_SIZE_MAX + constants.MAX_QUEUE_SIZE_MIN + <= parsed.max_queue_size + <= constants.MAX_QUEUE_SIZE_MAX ): return False, "max_queue_size out of range" + if parsed.max_event_age_sec <= 0: + return False, "max_event_age_sec must be > 0" + if parsed.max_dead_letter_batches < 0: + return False, "max_dead_letter_batches must be >= 0" return True, "OK" @@ -256,8 +321,10 @@ def doctor_report(parsed: argparse.Namespace) -> dict: integrations: list[dict[str, str]] = report["integrations"] # type: ignore[assignment] ok = True - dsn = parsed.dsn or os.environ.get("WILDEDGE_DSN") + client_env = read_client_env(dsn=parsed.dsn) + dsn = client_env.dsn + project_key = "default" if not dsn: ok = False checks.append( @@ -269,7 +336,7 @@ def doctor_report(parsed: argparse.Namespace) -> dict: ) else: try: - _, host_url = parse_dsn(dsn) + _, host_url, project_key = parse_dsn(dsn) checks.append({"name": "dsn", "status": "OK", "detail": host_url}) if parsed.network_check: reachable, detail = network_reachability_check(host_url) @@ -285,6 +352,17 @@ def doctor_report(parsed: argparse.Namespace) -> dict: ok = False checks.append({"name": "dsn", "status": "FAIL", "detail": str(exc)}) + app_identity = resolve_app_identity( + explicit=parsed.app_identity, + project_key=project_key, + ) + resolved_offline_queue_dir = parsed.offline_queue_dir or str( + default_pending_queue_dir(app_identity) + ) + resolved_dead_letter_dir = parsed.dead_letter_dir or str( + default_dead_letter_dir(app_identity) + ) + temp_ok, temp_detail = check_writable_dir(Path(tempfile.gettempdir())) checks.append( { @@ -315,6 +393,61 @@ def doctor_report(parsed: argparse.Namespace) -> dict: ) ok = ok and device_dir_ok + checks.append( + { + "name": "offline_queue_capacity", + "status": "OK", + "detail": f"max_queue_size={parsed.max_queue_size}; max_event_age_sec={parsed.max_event_age_sec}", + } + ) + checks.append( + { + "name": "dead_letter_capacity", + "status": "OK", + "detail": f"max_dead_letter_batches={parsed.max_dead_letter_batches}", + } + ) + + if parsed.offline_persistence: + offline_ok, offline_detail = check_writable_dir( + Path(resolved_offline_queue_dir) + ) + checks.append( + { + "name": "writable_offline_queue_dir", + "status": "OK" if offline_ok else "FAIL", + "detail": offline_detail, + } + ) + ok = ok and offline_ok + else: + checks.append( + { + "name": "writable_offline_queue_dir", + "status": "SKIP", + "detail": f"disabled ({resolved_offline_queue_dir})", + } + ) + + if parsed.dead_letter_persistence: + dl_ok, dl_detail = check_writable_dir(Path(resolved_dead_letter_dir)) + checks.append( + { + "name": "writable_dead_letter_dir", + "status": "OK" if dl_ok else "FAIL", + "detail": dl_detail, + } + ) + ok = ok and dl_ok + else: + checks.append( + { + "name": "writable_dead_letter_dir", + "status": "SKIP", + "detail": f"disabled ({resolved_dead_letter_dir})", + } + ) + for integration in integration_list(parsed.integrations): spec = INTEGRATIONS_BY_NAME.get(integration) if spec is None: diff --git a/wildedge/client.py b/wildedge/client.py index 9ba0e62..f4acd6f 100644 --- a/wildedge/client.py +++ b/wildedge/client.py @@ -1,15 +1,16 @@ from __future__ import annotations -import os import time import uuid import weakref +from collections.abc import Callable from datetime import datetime, timezone from typing import Any from urllib.parse import urlparse -from wildedge import config +from wildedge import constants from wildedge.consumer import Consumer +from wildedge.dead_letters import DeadLetterStore from wildedge.device import DeviceInfo, detect_device from wildedge.integrations.base import BaseExtractor from wildedge.integrations.gguf import GgufExtractor @@ -22,26 +23,32 @@ from wildedge.integrations.tensorflow import TensorflowExtractor from wildedge.logging import enable_debug, logger from wildedge.model import ModelHandle, ModelInfo, ModelRegistry +from wildedge.paths import ( + default_dead_letter_dir, + default_model_registry_path, + default_pending_queue_dir, +) from wildedge.queue import EventQueue, QueuePolicy +from wildedge.settings import read_client_env, resolve_app_identity from wildedge.timing import Timer, elapsed_ms from wildedge.transmitter import Transmitter DSN_FORMAT = "'https://@ingest.wildedge.dev/'" ERROR_DSN_MISSING_SECRET = f"DSN must include a project secret: {DSN_FORMAT}" ERROR_DSN_REQUIRED = ( - f"DSN is required. Pass dsn= or set {config.ENV_DSN}. Format: {DSN_FORMAT}" -) -ERROR_BATCH_SIZE_RANGE = ( - f"batch_size must be between {config.BATCH_SIZE_MIN} and {config.BATCH_SIZE_MAX}" + f"DSN is required. Pass dsn= or set {constants.ENV_DSN}. Format: {DSN_FORMAT}" ) +ERROR_BATCH_SIZE_RANGE = f"batch_size must be between {constants.BATCH_SIZE_MIN} and {constants.BATCH_SIZE_MAX}" ERROR_FLUSH_INTERVAL_RANGE = ( "flush_interval_sec must be between " - f"{config.FLUSH_INTERVAL_MIN} and {config.FLUSH_INTERVAL_MAX}" + f"{constants.FLUSH_INTERVAL_MIN} and {constants.FLUSH_INTERVAL_MAX}" ) ERROR_MAX_QUEUE_SIZE_RANGE = ( "max_queue_size must be between " - f"{config.MAX_QUEUE_SIZE_MIN} and {config.MAX_QUEUE_SIZE_MAX}" + f"{constants.MAX_QUEUE_SIZE_MIN} and {constants.MAX_QUEUE_SIZE_MAX}" ) +ERROR_MAX_EVENT_AGE = "max_event_age_sec must be greater than 0" +ERROR_MAX_DEAD_LETTER_BATCHES = "max_dead_letter_batches must be >= 0" ERROR_UNKNOWN_INTEGRATION = ( "Unknown integration {integration!r}. Available: {available}" ) @@ -52,15 +59,18 @@ ) -def parse_dsn(dsn: str) -> tuple[str, str]: - """Parse 'https://@ingest.wildedge.dev/' → (secret, host).""" +def parse_dsn(dsn: str) -> tuple[str, str, str]: + """Parse DSN into (secret, host, project_key).""" parsed = urlparse(dsn) if not parsed.username: raise ValueError(ERROR_DSN_MISSING_SECRET) + project_key = parsed.path.lstrip("/").split("/", 1)[0] + if not project_key: + raise ValueError(f"DSN must include a project key path: {DSN_FORMAT}") host = f"{parsed.scheme}://{parsed.hostname}" if parsed.port: host += f":{parsed.port}" - return parsed.username, host + return parsed.username, host, project_key DEFAULT_EXTRACTORS: list[BaseExtractor] = [ @@ -100,29 +110,51 @@ def __init__( app_version: str | None = None, device: DeviceInfo | None = None, queue_policy: QueuePolicy = QueuePolicy.OPPORTUNISTIC, - max_queue_size: int = config.DEFAULT_MAX_QUEUE_SIZE, - batch_size: int = config.DEFAULT_BATCH_SIZE, - flush_interval_sec: float = config.DEFAULT_FLUSH_INTERVAL_SEC, + max_queue_size: int = constants.DEFAULT_MAX_QUEUE_SIZE, + batch_size: int = constants.DEFAULT_BATCH_SIZE, + flush_interval_sec: float = constants.DEFAULT_FLUSH_INTERVAL_SEC, debug: bool | None = None, + max_event_age_sec: float = constants.DEFAULT_MAX_EVENT_AGE_SEC, + enable_offline_persistence: bool = constants.DEFAULT_ENABLE_OFFLINE_PERSISTENCE, + app_identity: str | None = None, + offline_queue_dir: str | None = None, + enable_dead_letter_persistence: bool = ( + constants.DEFAULT_ENABLE_DEAD_LETTER_PERSISTENCE + ), + dead_letter_dir: str | None = None, + max_dead_letter_batches: int = constants.DEFAULT_MAX_DEAD_LETTER_BATCHES, + on_delivery_failure: Callable[[str, int, int], None] | None = None, ): - dsn = dsn or os.environ.get(config.ENV_DSN) + env = read_client_env(dsn=dsn, debug=debug, app_identity=app_identity) + dsn = env.dsn if not dsn: raise ValueError(ERROR_DSN_REQUIRED) - api_key, host = parse_dsn(dsn) - if debug is None: - debug = os.environ.get(config.ENV_DEBUG, "").lower() in ("1", "true", "yes") + api_key, host, project_key = parse_dsn(dsn) + debug = env.debug + app_identity = resolve_app_identity( + explicit=env.app_identity, + project_key=project_key, + ) # Validate configuration ranges - if not (config.BATCH_SIZE_MIN <= batch_size <= config.BATCH_SIZE_MAX): + if not (constants.BATCH_SIZE_MIN <= batch_size <= constants.BATCH_SIZE_MAX): raise ValueError(ERROR_BATCH_SIZE_RANGE) if not ( - config.FLUSH_INTERVAL_MIN <= flush_interval_sec <= config.FLUSH_INTERVAL_MAX + constants.FLUSH_INTERVAL_MIN + <= flush_interval_sec + <= constants.FLUSH_INTERVAL_MAX ): raise ValueError(ERROR_FLUSH_INTERVAL_RANGE) if not ( - config.MAX_QUEUE_SIZE_MIN <= max_queue_size <= config.MAX_QUEUE_SIZE_MAX + constants.MAX_QUEUE_SIZE_MIN + <= max_queue_size + <= constants.MAX_QUEUE_SIZE_MAX ): raise ValueError(ERROR_MAX_QUEUE_SIZE_RANGE) + if max_event_age_sec <= 0: + raise ValueError(ERROR_MAX_EVENT_AGE) + if max_dead_letter_batches < 0: + raise ValueError(ERROR_MAX_DEAD_LETTER_BATCHES) self.api_key = api_key self.debug = debug @@ -135,9 +167,30 @@ def __init__( self.device = device or detect_device(api_key=api_key, app_version=app_version) self.session_id = str(uuid.uuid4()) self.created_at = datetime.now(timezone.utc) - self.queue = EventQueue(max_size=max_queue_size, policy=queue_policy) - self.registry = ModelRegistry() + resolved_offline_queue_dir = offline_queue_dir or str( + default_pending_queue_dir(app_identity) + ) + resolved_dead_letter_dir = dead_letter_dir or str( + default_dead_letter_dir(app_identity) + ) + resolved_model_registry_path = str(default_model_registry_path(app_identity)) + self.queue = EventQueue( + max_size=max_queue_size, + policy=queue_policy, + persist_to_disk=enable_offline_persistence, + disk_dir=resolved_offline_queue_dir, + ) + self.registry = ModelRegistry( + persist_path=resolved_model_registry_path + if enable_offline_persistence + else None + ) self.transmitter = Transmitter(api_key=api_key, host=host) + self.dead_letter_store = DeadLetterStore( + enabled=enable_dead_letter_persistence, + directory=resolved_dead_letter_dir, + max_batches=max_dead_letter_batches, + ) self.consumer = Consumer( queue=self.queue, transmitter=self.transmitter, @@ -147,6 +200,9 @@ def __init__( batch_size=batch_size, flush_interval_sec=flush_interval_sec, debug=debug, + max_event_age_sec=max_event_age_sec, + dead_letter_store=self.dead_letter_store, + on_delivery_failure=on_delivery_failure, ) self._auto_loaded: set[str] = set() @@ -165,6 +221,8 @@ def publish(self, event_dict: dict) -> None: event_dict.get("event_type"), event_dict.get("model_id"), ) + event_dict.setdefault("__we_first_queued_at", time.time()) + event_dict.setdefault("__we_attempts", 0) self.queue.add(event_dict) def register_model( @@ -425,10 +483,13 @@ def flush(self, timeout: float = 5.0) -> None: """Block until the event queue drains or timeout expires.""" self.consumer.flush(timeout=timeout) - def close(self) -> None: - """Flush remaining events and stop the consumer thread.""" + def close(self, timeout: float | None = None) -> None: + """Best-effort shutdown; pass timeout to attempt bounded flush first.""" self.closed = True - self.consumer.close() + if timeout is None: + self.consumer.close() + else: + self.consumer.close(timeout=timeout) def pending_count(self) -> int: """Return the number of events currently buffered.""" diff --git a/wildedge/config.py b/wildedge/constants.py similarity index 71% rename from wildedge/config.py rename to wildedge/constants.py index afd714b..5471767 100644 --- a/wildedge/config.py +++ b/wildedge/constants.py @@ -5,17 +5,24 @@ # Env vars ENV_DSN = "WILDEDGE_DSN" ENV_DEBUG = "WILDEDGE_DEBUG" +ENV_APP_IDENTITY = "WILDEDGE_APP_IDENTITY" # Defaults DEFAULT_MAX_QUEUE_SIZE = 200 DEFAULT_BATCH_SIZE = 10 DEFAULT_FLUSH_INTERVAL_SEC = 60.0 DEFAULT_HTTP_TIMEOUT = 15.0 +DEFAULT_MAX_EVENT_AGE_SEC = 15 * 60 +DEFAULT_ENABLE_OFFLINE_PERSISTENCE = True +DEFAULT_ENABLE_DEAD_LETTER_PERSISTENCE = False +DEFAULT_MAX_DEAD_LETTER_BATCHES = 10 +DEFAULT_SHUTDOWN_FLUSH_TIMEOUT_SEC = 0.0 # Consumer backoff BACKOFF_MIN = 1.0 BACKOFF_MAX = 60.0 BACKOFF_MULTIPLIER = 2.0 +BACKOFF_JITTER_RATIO = 0.2 IDLE_POLL_INTERVAL = 5.0 # Error message truncation diff --git a/wildedge/consumer.py b/wildedge/consumer.py index 6268757..7c33992 100644 --- a/wildedge/consumer.py +++ b/wildedge/consumer.py @@ -1,14 +1,16 @@ from __future__ import annotations import atexit +import random import threading import time from collections.abc import Callable from datetime import datetime, timezone -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Any -from wildedge import config +from wildedge import constants from wildedge.batch import build_batch +from wildedge.dead_letters import DeadLetterStore from wildedge.logging import logger from wildedge.queue import EventQueue from wildedge.transmitter import TransmitError, Transmitter @@ -30,6 +32,9 @@ def __init__( batch_size: int = 10, flush_interval_sec: float = 60.0, debug: bool = False, + max_event_age_sec: float = constants.DEFAULT_MAX_EVENT_AGE_SEC, + dead_letter_store: DeadLetterStore | None = None, + on_delivery_failure: Callable[[str, int, int], None] | None = None, ): self.queue = queue self.transmitter = transmitter @@ -39,17 +44,20 @@ def __init__( self.batch_size = batch_size self.flush_interval_sec = flush_interval_sec self.debug = debug + self.max_event_age_sec = max_event_age_sec + self.dead_letter_store = dead_letter_store + self.on_delivery_failure = on_delivery_failure self.stop_event = threading.Event() self.stopped = False - self.backoff = config.BACKOFF_MIN + self.backoff = constants.BACKOFF_MIN self.created_at = datetime.now(timezone.utc) self.thread = threading.Thread( target=self.run, daemon=True, name="wildedge-consumer" ) self.thread.start() - atexit.register(self.flush) + atexit.register(self.flush, constants.DEFAULT_SHUTDOWN_FLUSH_TIMEOUT_SEC) def run(self) -> None: last_flush = time.monotonic() @@ -62,20 +70,99 @@ def run(self) -> None: sent = self.drain_once() if sent: last_flush = time.monotonic() - self.backoff = config.BACKOFF_MIN + self.backoff = constants.BACKOFF_MIN else: - self.stop_event.wait(timeout=self.backoff) - self.backoff = min( - self.backoff * config.BACKOFF_MULTIPLIER, config.BACKOFF_MAX + wait_s, self.backoff = self.next_retry_delay( + self.backoff, + jitter=True, ) + self.stop_event.wait(timeout=wait_s) else: - self.stop_event.wait(timeout=config.IDLE_POLL_INTERVAL) + self.stop_event.wait(timeout=constants.IDLE_POLL_INTERVAL) + + def next_retry_delay( + self, + backoff: float, + *, + jitter: bool, + max_wait: float | None = None, + ) -> tuple[float, float]: + delay = backoff + if jitter: + delay += random.uniform(0, backoff * constants.BACKOFF_JITTER_RATIO) + if max_wait is not None: + delay = min(delay, max_wait) + next_backoff = min( + backoff * constants.BACKOFF_MULTIPLIER, constants.BACKOFF_MAX + ) + return delay, next_backoff + + def strip_internal_fields( + self, events: list[dict[str, Any]] + ) -> list[dict[str, Any]]: + return [ + {k: v for k, v in event.items() if not k.startswith("__we_")} + for event in events + ] + + def notify_delivery_failure(self, reason: str, dropped_count: int) -> None: + if self.on_delivery_failure is None: + return + try: + self.on_delivery_failure(reason, dropped_count, self.queue.length()) + except ( + Exception + ) as exc: # pragma: no cover - user callback failures are non-fatal + logger.warning("wildedge: on_delivery_failure callback failed: %s", exc) + + def dead_letter_and_drop( + self, + *, + reason: str, + events: list[dict[str, Any]], + batch_id: str | None = None, + details: dict[str, Any] | None = None, + ) -> None: + if self.dead_letter_store is not None: + self.dead_letter_store.write( + reason=reason, + events=self.strip_internal_fields(events), + batch_id=batch_id, + details=details, + ) + self.queue.remove_first_n(len(events)) + self.notify_delivery_failure(reason, len(events)) def drain_once(self) -> bool: events = self.queue.peek_many(self.batch_size) if not events: return False + now_unix = time.time() + expired_count = 0 + for event in events: + first_seen = float(event.get("__we_first_queued_at", now_unix)) + if (now_unix - first_seen) > self.max_event_age_sec: + expired_count += 1 + else: + break + if expired_count > 0: + expired = events[:expired_count] + self.dead_letter_and_drop( + reason="event_age_exceeded", + events=expired, + details={"max_event_age_sec": self.max_event_age_sec}, + ) + logger.warning( + "wildedge: dropped %d stale queued events (age > %.1fs)", + expired_count, + self.max_event_age_sec, + ) + return True + + for event in events: + event["__we_attempts"] = int(event.get("__we_attempts", 0)) + 1 + batch = build_batch( device=self.device, models=self.get_models(), @@ -107,8 +194,19 @@ def drain_once(self) -> bool: ) return True - if response.status in ("rejected", "unauthorized"): - self.queue.remove_first_n(len(events)) + # Permanent client/config errors should not be retried forever. + # Transmitter returns status="error" for non-retryable cases like 3xx/404. + if response.status in ("rejected", "unauthorized", "error"): + self.dead_letter_and_drop( + reason=f"permanent_{response.status}", + events=events, + batch_id=batch["batch_id"], + details={ + "response_status": response.status, + "events_accepted": response.events_accepted, + "events_rejected": response.events_rejected, + }, + ) return True return False @@ -118,10 +216,23 @@ def flush(self, timeout: float = 5.0) -> None: if self.stopped: return deadline = time.monotonic() + timeout + backoff = constants.BACKOFF_MIN while self.queue.length() > 0 and time.monotonic() < deadline: - self.drain_once() - if self.queue.length() > 0: - time.sleep(0.05) + progressed = self.drain_once() + if self.queue.length() == 0: + break + remaining = deadline - time.monotonic() + if remaining <= 0: + break + if progressed: + backoff = constants.BACKOFF_MIN + continue + sleep_for, backoff = self.next_retry_delay( + backoff, + jitter=True, + max_wait=remaining, + ) + time.sleep(sleep_for) def stop(self) -> None: """Signal the consumer to stop and wait for thread exit.""" @@ -129,7 +240,9 @@ def stop(self) -> None: self.stop_event.set() self.thread.join(timeout=2.0) - def close(self) -> None: - self.flush() + def close(self, timeout: float | None = None) -> None: + if timeout is None: + timeout = constants.DEFAULT_SHUTDOWN_FLUSH_TIMEOUT_SEC + self.flush(timeout=timeout) self.stop() self.transmitter.close() diff --git a/wildedge/dead_letters.py b/wildedge/dead_letters.py new file mode 100644 index 0000000..d583686 --- /dev/null +++ b/wildedge/dead_letters.py @@ -0,0 +1,70 @@ +from __future__ import annotations + +import json +import threading +import time +import uuid +from pathlib import Path +from typing import Any + +from wildedge.logging import logger + + +class DeadLetterStore: + """File-backed dead-letter store with capped batch count.""" + + def __init__( + self, + *, + enabled: bool, + directory: str, + max_batches: int, + ) -> None: + self.enabled = enabled + self.directory = Path(directory).expanduser() + self.max_batches = max_batches + self._lock = threading.Lock() + if self.enabled: + self.directory.mkdir(parents=True, exist_ok=True) + + def write( + self, + *, + reason: str, + events: list[dict[str, Any]], + batch_id: str | None = None, + details: dict[str, Any] | None = None, + ) -> None: + if not self.enabled: + return + payload = { + "id": str(uuid.uuid4()), + "at_unix": time.time(), + "reason": reason, + "batch_id": batch_id, + "event_count": len(events), + "details": details or {}, + "events": events, + } + data = json.dumps(payload, separators=(",", ":"), ensure_ascii=True).encode() + filename = f"{int(payload['at_unix'] * 1000)}-{payload['id']}.json" + path = self.directory / filename + with self._lock: + path.write_bytes(data) + self.enforce_limit() + + def enforce_limit(self) -> None: + if self.max_batches <= 0: + for path in sorted(self.directory.glob("*.json")): + path.unlink(missing_ok=True) + return + files = sorted(self.directory.glob("*.json")) + overflow = len(files) - self.max_batches + if overflow <= 0: + return + for path in files[:overflow]: + path.unlink(missing_ok=True) + logger.debug( + "wildedge: dead-letter cap enforced, removed %d old batch files", + overflow, + ) diff --git a/wildedge/decorators.py b/wildedge/decorators.py index 85c1706..fe2e3ff 100644 --- a/wildedge/decorators.py +++ b/wildedge/decorators.py @@ -4,7 +4,7 @@ import time from typing import TYPE_CHECKING, Any -from wildedge import config +from wildedge import constants from wildedge.timing import elapsed_ms if TYPE_CHECKING: @@ -68,7 +68,7 @@ def wrapper(*args, **kwargs): if self.capture_errors: self.handle.track_error( error_code="UNKNOWN", - error_message=str(exc)[: config.ERROR_MSG_MAX_LEN], + error_message=str(exc)[: constants.ERROR_MSG_MAX_LEN], ) raise @@ -86,7 +86,7 @@ def __exit__(self, exc_type, exc_val, exc_tb): if self.capture_errors: self.handle.track_error( error_code="UNKNOWN", - error_message=str(exc_val)[: config.ERROR_MSG_MAX_LEN] + error_message=str(exc_val)[: constants.ERROR_MSG_MAX_LEN] if exc_val else None, ) diff --git a/wildedge/device.py b/wildedge/device.py index 158410f..ed9b372 100644 --- a/wildedge/device.py +++ b/wildedge/device.py @@ -12,7 +12,7 @@ from datetime import datetime from pathlib import Path -from wildedge import config +from wildedge import constants from wildedge.platforms import CURRENT_PLATFORM from wildedge.platforms import PLATFORMS as _PLATFORMS from wildedge.platforms.base import debug_detection_failure @@ -45,7 +45,11 @@ def get_device_id_path() -> Path: """ Returns the path to the device ID file for the current platform. """ - return CURRENT_PLATFORM.config_base() / config.DEVICE_ID_DIR / config.DEVICE_ID_FILE + return ( + CURRENT_PLATFORM.config_base() + / constants.DEVICE_ID_DIR + / constants.DEVICE_ID_FILE + ) def load_or_create_device_uuid() -> str: @@ -105,7 +109,7 @@ def detect_timezone() -> str | None: class DeviceInfo: device_id: str device_type: str - sdk_version: str = config.SDK_VERSION + sdk_version: str = constants.SDK_VERSION device_model: str | None = None os_version: str | None = None locale: str | None = None diff --git a/wildedge/integrations/gguf.py b/wildedge/integrations/gguf.py index da41812..31d179e 100644 --- a/wildedge/integrations/gguf.py +++ b/wildedge/integrations/gguf.py @@ -8,7 +8,7 @@ from pathlib import Path from typing import TYPE_CHECKING -from wildedge import config +from wildedge import constants from wildedge.device import CURRENT_PLATFORM from wildedge.events.inference import GenerationOutputMeta, TextInputMeta from wildedge.integrations.base import BaseExtractor, patch_instance_call_once @@ -123,7 +123,7 @@ def patched_call(self_inner, *args, **kwargs): except Exception as exc: handle.track_error( error_code="UNKNOWN", - error_message=str(exc)[: config.ERROR_MSG_MAX_LEN], + error_message=str(exc)[: constants.ERROR_MSG_MAX_LEN], ) raise diff --git a/wildedge/integrations/keras.py b/wildedge/integrations/keras.py index 3883756..0e5bc4a 100644 --- a/wildedge/integrations/keras.py +++ b/wildedge/integrations/keras.py @@ -5,7 +5,7 @@ import time from typing import TYPE_CHECKING -from wildedge import config +from wildedge import constants from wildedge.device import CURRENT_PLATFORM from wildedge.integrations.base import BaseExtractor, patch_instance_call_once from wildedge.logging import logger @@ -48,7 +48,7 @@ def patched_call(self_inner, *args, **kwargs): except Exception as exc: handle.track_error( error_code="UNKNOWN", - error_message=str(exc)[: config.ERROR_MSG_MAX_LEN], + error_message=str(exc)[: constants.ERROR_MSG_MAX_LEN], ) raise diff --git a/wildedge/integrations/onnx.py b/wildedge/integrations/onnx.py index 9761840..43a3e34 100644 --- a/wildedge/integrations/onnx.py +++ b/wildedge/integrations/onnx.py @@ -7,7 +7,7 @@ import time from typing import TYPE_CHECKING, Any -from wildedge import config +from wildedge import constants from wildedge.events.inference import ( ClassificationOutputMeta, HistogramSummary, @@ -279,7 +279,7 @@ def patched_run(output_names, input_feed, run_options=None): except Exception as exc: handle.track_error( error_code="UNKNOWN", - error_message=str(exc)[: config.ERROR_MSG_MAX_LEN], + error_message=str(exc)[: constants.ERROR_MSG_MAX_LEN], ) raise diff --git a/wildedge/integrations/tensorflow.py b/wildedge/integrations/tensorflow.py index 89cc800..becfa5e 100644 --- a/wildedge/integrations/tensorflow.py +++ b/wildedge/integrations/tensorflow.py @@ -6,7 +6,7 @@ import time from typing import TYPE_CHECKING -from wildedge import config +from wildedge import constants from wildedge.device import CURRENT_PLATFORM from wildedge.integrations.base import BaseExtractor, patch_instance_call_once from wildedge.logging import logger @@ -72,7 +72,7 @@ def patched_call(self_inner, *args, **kwargs): except Exception as exc: handle.track_error( error_code="UNKNOWN", - error_message=str(exc)[: config.ERROR_MSG_MAX_LEN], + error_message=str(exc)[: constants.ERROR_MSG_MAX_LEN], ) raise @@ -105,7 +105,7 @@ def patched_predict(*args, **kwargs): except Exception as exc: handle.track_error( error_code="UNKNOWN", - error_message=str(exc)[: config.ERROR_MSG_MAX_LEN], + error_message=str(exc)[: constants.ERROR_MSG_MAX_LEN], ) raise diff --git a/wildedge/model.py b/wildedge/model.py index f9da8cb..5b66549 100644 --- a/wildedge/model.py +++ b/wildedge/model.py @@ -1,7 +1,9 @@ from __future__ import annotations +import json from collections.abc import Callable from dataclasses import dataclass +from pathlib import Path from typing import Any from wildedge.events import ( @@ -208,9 +210,45 @@ def track_error( class ModelRegistry: """Thread-safe registry mapping model_id to ModelInfo.""" - def __init__(self) -> None: + def __init__(self, persist_path: str | None = None) -> None: self.models: dict[str, ModelInfo] = {} self.handles: dict[str, ModelHandle] = {} + self.persist_path = Path(persist_path).expanduser() if persist_path else None + if self.persist_path is not None: + self.load_from_disk() + + def load_from_disk(self) -> None: + if self.persist_path is None or not self.persist_path.exists(): + return + try: + raw = json.loads(self.persist_path.read_text()) + except Exception: + return + if not isinstance(raw, dict): + return + for model_id, data in raw.items(): + if not isinstance(model_id, str) or not isinstance(data, dict): + continue + try: + self.models[model_id] = ModelInfo( + model_name=str(data["model_name"]), + model_version=str(data["model_version"]), + model_source=str(data["model_source"]), + model_format=str(data["model_format"]), + model_family=data.get("model_family"), + quantization=data.get("quantization"), + ) + except KeyError: + continue + + def save_to_disk(self) -> None: + if self.persist_path is None: + return + self.persist_path.parent.mkdir(parents=True, exist_ok=True) + payload = self.snapshot() + self.persist_path.write_text( + json.dumps(payload, separators=(",", ":"), ensure_ascii=True) + ) def register( self, model_id: str, info: ModelInfo, publish: Callable[[dict], None] @@ -218,8 +256,14 @@ def register( """Return (handle, is_new). is_new=False means already registered; skip install_hooks.""" if model_id in self.handles: return self.handles[model_id], False - handle = ModelHandle(model_id=model_id, info=info, publish=publish) - self.models[model_id] = info + + if model_id not in self.models: + self.models[model_id] = info + self.save_to_disk() + + handle = ModelHandle( + model_id=model_id, info=self.models[model_id], publish=publish + ) self.handles[model_id] = handle return handle, True diff --git a/wildedge/paths.py b/wildedge/paths.py new file mode 100644 index 0000000..de272e8 --- /dev/null +++ b/wildedge/paths.py @@ -0,0 +1,34 @@ +from __future__ import annotations + +import re +from pathlib import Path + +from wildedge.platforms import CURRENT_PLATFORM + + +def normalize_namespace(namespace: str) -> str: + cleaned = re.sub(r"[^A-Za-z0-9._-]+", "_", namespace.strip()) + cleaned = cleaned.strip("._-") + return cleaned or "default" + + +def default_sdk_state_dir() -> Path: + return CURRENT_PLATFORM.state_base() / "wildedge" + + +def default_sdk_cache_dir() -> Path: + return CURRENT_PLATFORM.cache_base() / "wildedge" + + +def default_pending_queue_dir(namespace: str = "default") -> Path: + return default_sdk_state_dir() / normalize_namespace(namespace) / "pending_queue" + + +def default_model_registry_path(namespace: str = "default") -> Path: + return ( + default_sdk_state_dir() / normalize_namespace(namespace) / "model_registry.json" + ) + + +def default_dead_letter_dir(namespace: str = "default") -> Path: + return default_sdk_cache_dir() / normalize_namespace(namespace) / "dead_letters" diff --git a/wildedge/platforms/base.py b/wildedge/platforms/base.py index 88ad775..28b060d 100644 --- a/wildedge/platforms/base.py +++ b/wildedge/platforms/base.py @@ -11,6 +11,8 @@ class PlatformAdapter(Protocol): wire_type: str def config_base(self) -> Path: ... + def state_base(self) -> Path: ... + def cache_base(self) -> Path: ... def device_model(self) -> str | None: ... diff --git a/wildedge/platforms/linux.py b/wildedge/platforms/linux.py index 04705a6..59a4834 100644 --- a/wildedge/platforms/linux.py +++ b/wildedge/platforms/linux.py @@ -1,5 +1,6 @@ from __future__ import annotations +import os import shutil from pathlib import Path @@ -17,6 +18,12 @@ class LinuxPlatform: def config_base(self) -> Path: return Path.home() / ".config" + def state_base(self) -> Path: + return Path(os.environ.get("XDG_STATE_HOME", Path.home() / ".local" / "state")) + + def cache_base(self) -> Path: + return Path(os.environ.get("XDG_CACHE_HOME", Path.home() / ".cache")) + def device_model(self) -> str | None: for path in ( "/sys/class/dmi/id/product_name", diff --git a/wildedge/platforms/macos.py b/wildedge/platforms/macos.py index 6b7e122..2d0313d 100644 --- a/wildedge/platforms/macos.py +++ b/wildedge/platforms/macos.py @@ -14,6 +14,12 @@ class MacOSPlatform: def config_base(self) -> Path: return Path.home() / ".config" + def state_base(self) -> Path: + return Path.home() / "Library" / "Application Support" + + def cache_base(self) -> Path: + return Path.home() / "Library" / "Caches" + def device_model(self) -> str | None: try: buf = ctypes.create_string_buffer(128) diff --git a/wildedge/platforms/unknown.py b/wildedge/platforms/unknown.py index 41a9f98..ddaa489 100644 --- a/wildedge/platforms/unknown.py +++ b/wildedge/platforms/unknown.py @@ -13,6 +13,12 @@ class UnknownPlatform: def config_base(self) -> Path: return Path.home() / ".config" + def state_base(self) -> Path: + return Path.home() / ".local" / "state" + + def cache_base(self) -> Path: + return Path.home() / ".cache" + def device_model(self) -> str | None: return None diff --git a/wildedge/platforms/windows.py b/wildedge/platforms/windows.py index 3ecce66..902bf01 100644 --- a/wildedge/platforms/windows.py +++ b/wildedge/platforms/windows.py @@ -23,6 +23,12 @@ class WindowsPlatform: def config_base(self) -> Path: return Path(os.environ.get("APPDATA", Path.home() / "AppData" / "Roaming")) + def state_base(self) -> Path: + return Path(os.environ.get("LOCALAPPDATA", Path.home() / "AppData" / "Local")) + + def cache_base(self) -> Path: + return self.state_base() + def device_model(self) -> str | None: if _winreg is None: return None diff --git a/wildedge/queue.py b/wildedge/queue.py index f580fcf..74c27f9 100644 --- a/wildedge/queue.py +++ b/wildedge/queue.py @@ -1,6 +1,10 @@ +import json import threading +import time +import uuid from collections import deque from enum import Enum +from pathlib import Path class QueuePolicy(Enum): @@ -18,12 +22,57 @@ class EventQueue: """Thread-safe FIFO queue with configurable full-queue policy.""" def __init__( - self, max_size: int = 200, policy: QueuePolicy = QueuePolicy.OPPORTUNISTIC + self, + max_size: int = 200, + policy: QueuePolicy = QueuePolicy.OPPORTUNISTIC, + persist_to_disk: bool = False, + disk_dir: str | None = None, ): self.max_size = max_size self.policy = policy self.events: deque[dict] = deque() + self.persist_to_disk = persist_to_disk + self.disk_dir = Path(disk_dir).expanduser() if disk_dir else None + self._event_paths: deque[Path] = deque() self.lock = threading.Lock() + if self.persist_to_disk: + if self.disk_dir is None: + raise ValueError("disk_dir is required when persist_to_disk=True") + self.disk_dir.mkdir(parents=True, exist_ok=True) + self.load_from_disk() + + def load_from_disk(self) -> None: + assert self.disk_dir is not None + files = sorted(self.disk_dir.glob("*.json")) + for path in files: + try: + payload = json.loads(path.read_text()) + if isinstance(payload, dict): + self.events.append(payload) + self._event_paths.append(path) + else: + path.unlink(missing_ok=True) + except Exception: + path.unlink(missing_ok=True) + while len(self.events) > self.max_size: + self.events.popleft() + dropped = self._event_paths.popleft() + dropped.unlink(missing_ok=True) + + def persist_event(self, event: dict) -> None: + if not self.persist_to_disk: + return + assert self.disk_dir is not None + filename = f"{time.time_ns()}-{uuid.uuid4()}.json" + path = self.disk_dir / filename + path.write_text(json.dumps(event, separators=(",", ":"), ensure_ascii=True)) + self._event_paths.append(path) + + def drop_oldest_for_capacity(self) -> None: + self.events.popleft() + if self.persist_to_disk and self._event_paths: + path = self._event_paths.popleft() + path.unlink(missing_ok=True) def add(self, event: dict) -> None: with self.lock: @@ -33,8 +82,8 @@ def add(self, event: dict) -> None: f"Event queue is full ({self.max_size} items). " "Use QueuePolicy.OPPORTUNISTIC to silently drop oldest." ) - # OPPORTUNISTIC: drop oldest to make room - self.events.popleft() + self.drop_oldest_for_capacity() + self.persist_event(event) self.events.append(event) def peek(self) -> dict | None: @@ -49,11 +98,17 @@ def remove_first(self) -> None: with self.lock: if self.events: self.events.popleft() + if self.persist_to_disk and self._event_paths: + path = self._event_paths.popleft() + path.unlink(missing_ok=True) def remove_first_n(self, n: int) -> None: with self.lock: for _ in range(min(n, len(self.events))): self.events.popleft() + if self.persist_to_disk and self._event_paths: + path = self._event_paths.popleft() + path.unlink(missing_ok=True) def length(self) -> int: with self.lock: diff --git a/wildedge/runtime/bootstrap.py b/wildedge/runtime/bootstrap.py index 4d66704..c2b3fb5 100644 --- a/wildedge/runtime/bootstrap.py +++ b/wildedge/runtime/bootstrap.py @@ -12,17 +12,19 @@ from importlib import metadata from wildedge.client import WildEdge -from wildedge.config import ENV_DSN +from wildedge.constants import ENV_DSN from wildedge.integrations.registry import INTEGRATIONS_BY_NAME, supported_integrations - -RUN_DSN_ENV = "WILDEDGE_RUN_DSN" -RUN_APP_VERSION_ENV = "WILDEDGE_RUN_APP_VERSION" -RUN_DEBUG_ENV = "WILDEDGE_RUN_DEBUG" -RUN_FLUSH_TIMEOUT_ENV = "WILDEDGE_RUN_FLUSH_TIMEOUT" -RUN_INTEGRATIONS_ENV = "WILDEDGE_RUN_INTEGRATIONS" -RUN_STRICT_INTEGRATIONS_ENV = "WILDEDGE_RUN_STRICT_INTEGRATIONS" -RUN_PROPAGATE_ENV = "WILDEDGE_RUN_PROPAGATE" -RUN_PRINT_STARTUP_REPORT_ENV = "WILDEDGE_RUN_PRINT_STARTUP_REPORT" +from wildedge.settings import ( + RUN_APP_VERSION_ENV, + RUN_DEBUG_ENV, + RUN_DSN_ENV, + RUN_FLUSH_TIMEOUT_ENV, + RUN_INTEGRATIONS_ENV, + RUN_PRINT_STARTUP_REPORT_ENV, + RUN_PROPAGATE_ENV, + RUN_STRICT_INTEGRATIONS_ENV, + read_runtime_env, +) SUPPORTED_SIGNALS = [signal.SIGINT, signal.SIGTERM] STATUS_OK_PATCHED = "OK_PATCHED" @@ -32,16 +34,6 @@ STRICT_FAILURE_STATUSES = {STATUS_SKIP_MISSING_DEP, STATUS_ERROR_PATCH_FAILED} -def _as_bool(value: str | None) -> bool: - return (value or "").strip().lower() in {"1", "true", "yes", "on"} - - -def _integration_list(value: str | None) -> list[str]: - if not value or value == "all": - return sorted(supported_integrations()) - return [item.strip() for item in value.split(",") if item.strip()] - - def clear_runtime_env() -> None: """Remove run-scoped env vars so nested processes do not inherit runtime config.""" for key in ( @@ -110,23 +102,18 @@ def format_startup_report(context: RuntimeContext) -> str: def install_runtime() -> RuntimeContext: """Create and configure WildEdge client for process-level instrumentation.""" - dsn = os.environ.get(RUN_DSN_ENV) or os.environ.get(ENV_DSN) - if not dsn: - raise RuntimeConfigError( - f"{ENV_DSN} (or {RUN_DSN_ENV}) must be set to use `wildedge run`." - ) - - app_version = os.environ.get(RUN_APP_VERSION_ENV) - debug = _as_bool(os.environ.get(RUN_DEBUG_ENV)) - print_startup_report = _as_bool(os.environ.get(RUN_PRINT_STARTUP_REPORT_ENV)) - strict_integrations = _as_bool(os.environ.get(RUN_STRICT_INTEGRATIONS_ENV)) try: - flush_timeout = float(os.environ.get(RUN_FLUSH_TIMEOUT_ENV, "5.0")) + env = read_runtime_env(all_integrations=sorted(supported_integrations())) except ValueError as exc: raise RuntimeConfigError("invalid flush timeout") from exc - client = WildEdge(dsn=dsn, app_version=app_version, debug=debug) - integrations = _integration_list(os.environ.get(RUN_INTEGRATIONS_ENV)) + if not env.dsn: + raise RuntimeConfigError( + f"{ENV_DSN} (or {RUN_DSN_ENV}) must be set to use `wildedge run`." + ) + + client = WildEdge(dsn=env.dsn, app_version=env.app_version, debug=env.debug) + integrations = env.integrations statuses: list[dict[str, str]] = [] for integration in integrations: spec = INTEGRATIONS_BY_NAME.get(integration) @@ -165,13 +152,13 @@ def install_runtime() -> RuntimeContext: "detail": str(exc), } ) - if debug: + if env.debug: print( f"wildedge: instrument({integration!r}) failed: {exc}", file=os.sys.stderr, ) - if strict_integrations: + if env.strict_integrations: failures = [row for row in statuses if row["status"] in STRICT_FAILURE_STATUSES] if failures: fail_detail = ", ".join( @@ -183,9 +170,9 @@ def install_runtime() -> RuntimeContext: context = RuntimeContext( client=client, - flush_timeout=flush_timeout, - debug=debug, - print_startup_report=print_startup_report, + flush_timeout=env.flush_timeout, + debug=env.debug, + print_startup_report=env.print_startup_report, integration_statuses=statuses, ) atexit.register(context.shutdown) diff --git a/wildedge/runtime/runner.py b/wildedge/runtime/runner.py index 3ad28d1..bda0009 100644 --- a/wildedge/runtime/runner.py +++ b/wildedge/runtime/runner.py @@ -3,20 +3,17 @@ from __future__ import annotations import argparse -import os import runpy import sys from wildedge.runtime.bootstrap import ( - RUN_PRINT_STARTUP_REPORT_ENV, - RUN_PROPAGATE_ENV, RuntimeConfigError, RuntimeStrictIntegrationError, - _as_bool, clear_runtime_env, format_startup_report, install_runtime, ) +from wildedge.settings import read_runner_env EXIT_CONFIG_ERROR = 120 EXIT_STRICT_INTEGRATION_ERROR = 121 @@ -50,14 +47,15 @@ def main(argv: list[str] | None = None) -> int: print(f"wildedge: bootstrap internal error: {exc}", file=sys.stderr) return EXIT_BOOTSTRAP_INTERNAL_ERROR + runner_env = read_runner_env() if ( getattr(context, "debug", False) or getattr(context, "print_startup_report", False) - or _as_bool(os.environ.get(RUN_PRINT_STARTUP_REPORT_ENV)) + or runner_env.print_startup_report ): print(format_startup_report(context), file=sys.stderr) - if not _as_bool(os.environ.get(RUN_PROPAGATE_ENV, "1")): + if not runner_env.propagate: clear_runtime_env() try: if parsed.mode == "script": diff --git a/wildedge/settings.py b/wildedge/settings.py new file mode 100644 index 0000000..331a8c4 --- /dev/null +++ b/wildedge/settings.py @@ -0,0 +1,120 @@ +from __future__ import annotations + +import os +from collections.abc import Mapping +from dataclasses import dataclass + +from wildedge import constants + +RUN_DSN_ENV = "WILDEDGE_RUN_DSN" +RUN_APP_VERSION_ENV = "WILDEDGE_RUN_APP_VERSION" +RUN_DEBUG_ENV = "WILDEDGE_RUN_DEBUG" +RUN_FLUSH_TIMEOUT_ENV = "WILDEDGE_RUN_FLUSH_TIMEOUT" +RUN_INTEGRATIONS_ENV = "WILDEDGE_RUN_INTEGRATIONS" +RUN_STRICT_INTEGRATIONS_ENV = "WILDEDGE_RUN_STRICT_INTEGRATIONS" +RUN_PROPAGATE_ENV = "WILDEDGE_RUN_PROPAGATE" +RUN_PRINT_STARTUP_REPORT_ENV = "WILDEDGE_RUN_PRINT_STARTUP_REPORT" + +TRUE_VALUES = {"1", "true", "yes", "on"} + + +@dataclass(frozen=True) +class ClientEnv: + dsn: str | None + debug: bool + app_identity: str | None + + +@dataclass(frozen=True) +class RuntimeEnv: + dsn: str | None + app_version: str | None + debug: bool + print_startup_report: bool + strict_integrations: bool + flush_timeout: float + integrations: list[str] + propagate: bool + + +@dataclass(frozen=True) +class RunnerEnv: + print_startup_report: bool + propagate: bool + + +def parse_bool(value: str | None) -> bool: + return (value or "").strip().lower() in TRUE_VALUES + + +def parse_integration_list(value: str | None, all_values: list[str]) -> list[str]: + if not value or value == "all": + return sorted(all_values) + return [item.strip() for item in value.split(",") if item.strip()] + + +def resolve_app_identity( + *, + explicit: str | None, + project_key: str, + environ: Mapping[str, str] | None = None, +) -> str: + env = environ if environ is not None else os.environ + return explicit or env.get(constants.ENV_APP_IDENTITY) or project_key + + +def read_client_env( + *, + dsn: str | None = None, + debug: bool | None = None, + app_identity: str | None = None, + environ: Mapping[str, str] | None = None, +) -> ClientEnv: + env = environ if environ is not None else os.environ + resolved_dsn = dsn or env.get(constants.ENV_DSN) + resolved_debug = ( + debug if debug is not None else parse_bool(env.get(constants.ENV_DEBUG)) + ) + resolved_identity = app_identity or env.get(constants.ENV_APP_IDENTITY) + return ClientEnv( + dsn=resolved_dsn, + debug=resolved_debug, + app_identity=resolved_identity, + ) + + +def read_runtime_env( + *, + all_integrations: list[str], + environ: Mapping[str, str] | None = None, +) -> RuntimeEnv: + env = environ if environ is not None else os.environ + flush_timeout = float( + env.get( + RUN_FLUSH_TIMEOUT_ENV, + str(constants.DEFAULT_SHUTDOWN_FLUSH_TIMEOUT_SEC), + ) + ) + return RuntimeEnv( + dsn=env.get(RUN_DSN_ENV) or env.get(constants.ENV_DSN), + app_version=env.get(RUN_APP_VERSION_ENV), + debug=parse_bool(env.get(RUN_DEBUG_ENV)), + print_startup_report=parse_bool(env.get(RUN_PRINT_STARTUP_REPORT_ENV)), + strict_integrations=parse_bool(env.get(RUN_STRICT_INTEGRATIONS_ENV)), + flush_timeout=flush_timeout, + integrations=parse_integration_list( + env.get(RUN_INTEGRATIONS_ENV), all_integrations + ), + propagate=parse_bool(env.get(RUN_PROPAGATE_ENV, "1")), + ) + + +def read_runner_env( + *, + environ: Mapping[str, str] | None = None, +) -> RunnerEnv: + env = environ if environ is not None else os.environ + return RunnerEnv( + print_startup_report=parse_bool(env.get(RUN_PRINT_STARTUP_REPORT_ENV)), + propagate=parse_bool(env.get(RUN_PROPAGATE_ENV, "1")), + ) diff --git a/wildedge/transmitter.py b/wildedge/transmitter.py index c58bf62..c9e2966 100644 --- a/wildedge/transmitter.py +++ b/wildedge/transmitter.py @@ -8,7 +8,7 @@ import urllib.request from dataclasses import dataclass -from wildedge import config +from wildedge import constants from wildedge.logging import logger @@ -37,12 +37,12 @@ class Transmitter: """Sends batch envelopes to the WildEdge ingest endpoint (stdlib urllib).""" def __init__( - self, api_key: str, host: str, timeout: float = config.DEFAULT_HTTP_TIMEOUT + self, api_key: str, host: str, timeout: float = constants.DEFAULT_HTTP_TIMEOUT ): self.host = host.rstrip("/") self.timeout = timeout self.headers = { - "User-Agent": config.SDK_VERSION, + "User-Agent": constants.SDK_VERSION, "X-Project-Secret": api_key, "Content-Type": "application/json", } @@ -82,7 +82,7 @@ def send(self, batch: dict) -> IngestResponse: if status_code == 400: logger.warning( "wildedge: batch rejected (400) - discarding: %s", - raw[: config.ERROR_MSG_MAX_LEN], + raw[: constants.ERROR_MSG_MAX_LEN], ) return IngestResponse( status="rejected", @@ -106,7 +106,7 @@ def send(self, batch: dict) -> IngestResponse: logger.error( "wildedge: unexpected redirect (%d) to %s; check WILDEDGE_DSN", status_code, - raw[: config.ERROR_MSG_MAX_LEN], + raw[: constants.ERROR_MSG_MAX_LEN], ) return IngestResponse( status="error", @@ -128,7 +128,7 @@ def send(self, batch: dict) -> IngestResponse: if status_code == 429 or status_code >= 500: raise TransmitError( - f"HTTP {status_code}: {raw[: config.ERROR_MSG_MAX_LEN]!r}" + f"HTTP {status_code}: {raw[: constants.ERROR_MSG_MAX_LEN]!r}" ) if 400 <= status_code < 500: @@ -136,7 +136,7 @@ def send(self, batch: dict) -> IngestResponse: logger.warning( "wildedge: batch rejected (%d) - discarding: %s", status_code, - raw[: config.ERROR_MSG_MAX_LEN], + raw[: constants.ERROR_MSG_MAX_LEN], ) return IngestResponse( status="rejected",