more timeouts

Author: CamDavidsonPilon

CHANGELOG.md

@@ -33,6 +33,7 @@
  - Fix self-test logging closing prematurely.
  - Fix floating point error at the boundary of OD calibrations.
  - Fix runtime forward-reference errors in type annotations after dropping `__future__` imports.
+ - Fix timeouts being too short on some UI export operations
 
 ### 25.12.10
 

core/pioreactor/web/api.py

@@ -110,20 +110,23 @@ def broadcast_get_across_cluster(endpoint: str, timeout: float = 5.0, return_raw
 
 
 def broadcast_post_across_cluster(
-    endpoint: str, json: dict | None = None, params: dict | None = None
+    endpoint: str,
+    json: dict | None = None,
+    params: dict | None = None,
+    timeout: float = 30.0,
 ) -> Result:
     assert endpoint.startswith("/unit_api")
-    return tasks.multicast_post(endpoint, get_all_units(), json=json, params=params)
+    return tasks.multicast_post(endpoint, get_all_units(), json=json, params=params, timeout=timeout)
 
 
-def broadcast_delete_across_cluster(endpoint: str, json: dict | None = None) -> Result:
+def broadcast_delete_across_cluster(endpoint: str, json: dict | None = None, timeout: float = 30.0) -> Result:
     assert endpoint.startswith("/unit_api")
-    return tasks.multicast_delete(endpoint, get_all_units(), json=json)
+    return tasks.multicast_delete(endpoint, get_all_units(), json=json, timeout=timeout)
 
 
-def broadcast_patch_across_cluster(endpoint: str, json: dict | None = None) -> Result:
+def broadcast_patch_across_cluster(endpoint: str, json: dict | None = None, timeout: float = 30.0) -> Result:
     assert endpoint.startswith("/unit_api")
-    return tasks.multicast_patch(endpoint, get_all_units(), json=json)
+    return tasks.multicast_patch(endpoint, get_all_units(), json=json, timeout=timeout)
 
 
 # send only to workers
@@ -147,20 +150,23 @@ def broadcast_get_across_workers_in_experiment(
 
 
 def broadcast_post_across_workers(
-    endpoint: str, json: dict | None = None, params: dict | None = None
+    endpoint: str,
+    json: dict | None = None,
+    params: dict | None = None,
+    timeout: float = 30.0,
 ) -> Result:
     assert endpoint.startswith("/unit_api")
-    return tasks.multicast_post(endpoint, get_all_workers(), json=json, params=params)
+    return tasks.multicast_post(endpoint, get_all_workers(), json=json, params=params, timeout=timeout)
 
 
-def broadcast_delete_across_workers(endpoint: str, json: dict | None = None) -> Result:
+def broadcast_delete_across_workers(endpoint: str, json: dict | None = None, timeout: float = 30.0) -> Result:
     assert endpoint.startswith("/unit_api")
-    return tasks.multicast_delete(endpoint, get_all_workers(), json=json)
+    return tasks.multicast_delete(endpoint, get_all_workers(), json=json, timeout=timeout)
 
 
-def broadcast_patch_across_workers(endpoint: str, json: dict | None = None) -> Result:
+def broadcast_patch_across_workers(endpoint: str, json: dict | None = None, timeout: float = 30.0) -> Result:
     assert endpoint.startswith("/unit_api")
-    return tasks.multicast_patch(endpoint, get_all_workers(), json=json)
+    return tasks.multicast_patch(endpoint, get_all_workers(), json=json, timeout=timeout)
 
 
 def _build_single_file_multipart(
@@ -1536,17 +1542,15 @@ def start_calibration_session(pioreactor_unit: str) -> ResponseReturnValue:
             timeout=30,
         )
         response.raise_for_status()
-    except (HTTPErrorStatus, HTTPException) as exc:
+    except (HTTPErrorStatus, HTTPException):
         detail = _extract_unit_api_error(response)
         if detail:
-            publish_to_error_log(f"{exc}: {detail}", "start_calibration_session")
             abort_with(502, f"{detail}")
         if response is not None:
             abort_with(
                 502,
                 f"Starting calibration session failed on {pioreactor_unit} (HTTP {response.status_code}).",
             )
-        publish_to_error_log(str(exc), "start_calibration_session")
         abort_with(502, f"Starting calibration session failed on {pioreactor_unit}.")
 
     return Response(
@@ -1569,17 +1573,15 @@ def get_calibration_session(pioreactor_unit: str, session_id: str) -> ResponseRe
             timeout=30,
         )
         response.raise_for_status()
-    except (HTTPErrorStatus, HTTPException) as exc:
+    except (HTTPErrorStatus, HTTPException):
         detail = _extract_unit_api_error(response)
         if detail:
-            publish_to_error_log(f"{exc}: {detail}", "get_calibration_session")
             abort_with(502, f"Fetching calibration session failed on {pioreactor_unit}: {detail}")
         if response is not None:
             abort_with(
                 502,
                 f"Fetching calibration session failed on {pioreactor_unit} (HTTP {response.status_code}).",
             )
-        publish_to_error_log(str(exc), "get_calibration_session")
         abort_with(502, f"Fetching calibration session failed on {pioreactor_unit}.")
 
     return Response(
@@ -1607,17 +1609,15 @@ def advance_calibration_session(pioreactor_unit: str, session_id: str) -> Respon
             timeout=300,
         )
         response.raise_for_status()
-    except (HTTPErrorStatus, HTTPException) as exc:
+    except (HTTPErrorStatus, HTTPException):
         detail = _extract_unit_api_error(response)
         if detail:
-            publish_to_error_log(f"{exc}: {detail}", "advance_calibration_session")
             abort_with(502, detail)
         if response is not None:
             abort_with(
                 502,
                 f"Updating calibration session failed on {pioreactor_unit} (HTTP {response.status_code}).",
             )
-        publish_to_error_log(str(exc), "advance_calibration_session")
         abort_with(502, f"Updating calibration session failed on {pioreactor_unit}.")
 
     return Response(
@@ -1640,17 +1640,15 @@ def abort_calibration_session(pioreactor_unit: str, session_id: str) -> Response
             timeout=30,
         )
         response.raise_for_status()
-    except (HTTPErrorStatus, HTTPException) as exc:
+    except (HTTPErrorStatus, HTTPException):
         detail = _extract_unit_api_error(response)
         if detail:
-            publish_to_error_log(f"{exc}: {detail}", "abort_calibration_session")
             abort_with(502, f"Aborting calibration session failed on {pioreactor_unit}: {detail}")
         if response is not None:
             abort_with(
                 502,
                 f"Aborting calibration session failed on {pioreactor_unit} (HTTP {response.status_code}).",
             )
-        publish_to_error_log(str(exc), "abort_calibration_session")
         abort_with(502, f"Aborting calibration session failed on {pioreactor_unit}.")
 
     return Response(

core/pioreactor/web/tasks.py

@@ -890,6 +890,7 @@ def multicast_post(
     units: list[str],
     json: dict | list[dict | None] | None = None,
     params: dict | list[dict | None] | None = None,
+    timeout: float = 30.0,
 ) -> dict[str, Any]:
     # this function "consumes" one huey thread waiting fyi
     assert endpoint.startswith("/unit_api")
@@ -905,7 +906,7 @@ def multicast_post(
     tasks = post_into_unit.map(((units[i], endpoint, json[i], params[i]) for i in range(len(units))))
 
     return {
-        unit: response for (unit, response) in tasks.get(blocking=True, timeout=30)
+        unit: response for (unit, response) in tasks.get(blocking=True, timeout=timeout)
     }  # add a timeout so that we don't hold up a thread forever.
 
 
@@ -964,7 +965,7 @@ def multicast_get(
 
     tasks = get_from_unit.map(((units[i], endpoint, json[i], timeout, return_raw) for i in range(len(units))))
     unsorted_responses = {
-        unit: response for (unit, response) in tasks.get(blocking=True, timeout=15)
+        unit: response for (unit, response) in tasks.get(blocking=True, timeout=timeout)
     }  # add a timeout so that we don't hold up a thread forever.
 
     return dict(sorted(unsorted_responses.items()))  # always sort alphabetically for downstream uses.
@@ -996,14 +997,16 @@ def patch_into_unit(unit: str, endpoint: str, json: dict | None = None) -> tuple
 
 
 @huey.task(priority=50)
-def multicast_patch(endpoint: str, units: list[str], json: dict | None = None) -> dict[str, Any]:
+def multicast_patch(
+    endpoint: str, units: list[str], json: dict | None = None, timeout: float = 30.0
+) -> dict[str, Any]:
     # this function "consumes" one huey thread waiting fyi
     assert endpoint.startswith("/unit_api")
 
     tasks = patch_into_unit.map(((unit, endpoint, json) for unit in units))
 
     return {
-        unit: response for (unit, response) in tasks.get(blocking=True, timeout=30)
+        unit: response for (unit, response) in tasks.get(blocking=True, timeout=timeout)
     }  # add a timeout so that we don't hold up a thread forever.
 
 
@@ -1026,12 +1029,14 @@ def delete_from_unit(unit: str, endpoint: str, json: dict | None = None) -> tupl
 
 
 @huey.task(priority=5)
-def multicast_delete(endpoint: str, units: list[str], json: dict | None = None) -> dict[str, Any]:
+def multicast_delete(
+    endpoint: str, units: list[str], json: dict | None = None, timeout: float = 30.0
+) -> dict[str, Any]:
     # this function "consumes" one huey thread waiting fyi
     assert endpoint.startswith("/unit_api")
 
     tasks = delete_from_unit.map(((unit, endpoint, json) for unit in units))
 
     return {
-        unit: response for (unit, response) in tasks.get(blocking=True, timeout=30)
+        unit: response for (unit, response) in tasks.get(blocking=True, timeout=timeout)
     }  # add a timeout so that we don't hold up a thread forever.