avoid a worker being held up indefintely by constantly checking

Author: CamDavidsonPilon

core/pioreactor/web/tasks.py

@@ -163,7 +163,7 @@ def _process_delayed_json_response(
     unit: str,
     response: Response,
     *,
-    max_attempts: int = 50,
+    max_attempts: int = 60,
     retry_sleep_s: float = 0.1,
 ) -> tuple[str, Any]:
     """
@@ -801,7 +801,7 @@ def _get_from_unit(
     json: dict | None = None,
     timeout=5.0,
     return_raw=False,
-    max_attempts: int = 30,
+    max_attempts: int = 60,
 ) -> tuple[str, Any]:
     try:
         address = resolve_to_address(unit)

core/tests/web/test_tasks.py

@@ -0,0 +1,64 @@
+# -*- coding: utf-8 -*-
+from __future__ import annotations
+
+import json
+from typing import Any
+
+import pytest
+from pioreactor.mureq import Response
+from pioreactor.web import tasks
+
+
+def _response(status_code: int, payload: dict[str, Any]) -> Response:
+    return Response("http://unit.local", status_code, {}, json.dumps(payload).encode())
+
+
+def test_get_from_unit_retries_until_result(monkeypatch: pytest.MonkeyPatch) -> None:
+    # Simulate two pending responses followed by a completed task.
+    responses = [
+        _response(202, {"result_url_path": "/unit_api/task_results/abc"}),
+        _response(202, {"result_url_path": "/unit_api/task_results/abc"}),
+        _response(200, {"task_id": "abc", "result": {"ok": True}}),
+    ]
+
+    # Each request pops the next response in sequence.
+    def fake_get_from(
+        address: str, endpoint: str, json: dict | None = None, timeout: float = 5.0
+    ) -> Response:
+        return responses.pop(0)
+
+    monkeypatch.setattr(tasks, "get_from", fake_get_from)
+    monkeypatch.setattr(tasks, "resolve_to_address", lambda unit: "http://unit.local")
+    # Avoid test delays from retry sleeps.
+    monkeypatch.setattr(tasks, "sleep", lambda _: None)
+
+    unit, result = tasks._get_from_unit("unit1", "/unit_api/do", max_attempts=2)
+
+    assert unit == "unit1"
+    assert result == {"ok": True}
+    assert responses == []
+
+
+def test_get_from_unit_stops_after_max_attempts(monkeypatch: pytest.MonkeyPatch) -> None:
+    # Simulate a pending response that never resolves within the attempt limit.
+    responses = [
+        _response(202, {"result_url_path": "/unit_api/task_results/abc"}),
+        _response(202, {"result_url_path": "/unit_api/task_results/abc"}),
+    ]
+
+    # Each request pops the next response in sequence.
+    def fake_get_from(
+        address: str, endpoint: str, json: dict | None = None, timeout: float = 5.0
+    ) -> Response:
+        return responses.pop(0)
+
+    monkeypatch.setattr(tasks, "get_from", fake_get_from)
+    monkeypatch.setattr(tasks, "resolve_to_address", lambda unit: "http://unit.local")
+    # Avoid test delays from retry sleeps.
+    monkeypatch.setattr(tasks, "sleep", lambda _: None)
+
+    unit, result = tasks._get_from_unit("unit1", "/unit_api/do", max_attempts=1)
+
+    assert unit == "unit1"
+    assert result is None
+    assert responses == []