Offline mode (#4)

Author: piotrekno1

README.md

@@ -191,10 +191,17 @@ with wildedge.track(handle):
 |---|---|---|---|
 | `dsn` | `-` | `WILDEDGE_DSN` | Required. `https://<secret>@ingest.wildedge.dev/<key>` |
 | `app_version` | `None` | `-` | Optional. Your app's version string. |
+| `app_identity` | `<project_key>` | `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
 

scripts/run_compat_local.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)

tests/test_batch.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

tests/test_cli.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")