Additional hardening for transient exceptions, fix intercepted exception (try/finally instead of try/catch/reraise)

Author: chrishagglund-ship-it

tests/integration/test_async_lease_extension.py

@@ -38,12 +38,39 @@
 from conductor.client.http.models.task_def import TaskDef
 from conductor.client.http.models.workflow_task import WorkflowTask
 from conductor.client.http.models.start_workflow_request import StartWorkflowRequest
+from conductor.client.http.rest import ApiException
 from conductor.client.orkes.orkes_workflow_client import OrkesWorkflowClient
 from conductor.client.orkes.orkes_metadata_client import OrkesMetadataClient
 
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
+
+def _is_transient(e):
+    """A transient (status 0) ApiException is a raw transport blip against the
+    shared dev server (read timeout, connection reset, HTTP/2 GOAWAY, stale
+    keep-alive) — not a real failure. status 0 means no HTTP response was
+    received at all."""
+    return isinstance(e, ApiException) and (
+        getattr(e, 'transient', False) or e.status in (0, None))
+
+
+def _retry_on_transient(func, *args, retries=4, base_delay=1, **kwargs):
+    """Retry a workflow/metadata client call on a transient (0) transport blip.
+    These `(0)` errors flake the lease suites on a loaded shared server; retrying
+    absorbs the noise. Non-transient errors (real 4xx/5xx) raise immediately."""
+    for attempt in range(retries):
+        try:
+            return func(*args, **kwargs)
+        except ApiException as e:
+            if _is_transient(e) and attempt < retries - 1:
+                logger.warning(
+                    'transient (%s) API error (attempt %d/%d): %s; retrying',
+                    e.status, attempt + 1, retries, e)
+                time.sleep(base_delay * (2 ** attempt))
+                continue
+            raise
+
 # Short response timeout — task must heartbeat to stay alive
 RESPONSE_TIMEOUT_SECONDS = 10
 
@@ -220,9 +247,11 @@ def _register_workflow(self, wf_name, task_names):
             ))
         workflow._tasks = tasks
         try:
-            self.metadata_client.update_workflow_def(workflow, overwrite=True)
+            _retry_on_transient(self.metadata_client.update_workflow_def,
+                                 workflow, overwrite=True)
         except Exception:
-            self.metadata_client.register_workflow_def(workflow, overwrite=True)
+            _retry_on_transient(self.metadata_client.register_workflow_def,
+                                 workflow, overwrite=True)
         logger.info("Registered workflow: %s", wf_name)
 
     def _start_workflow(self, wf_name, job_id):
@@ -231,7 +260,8 @@ def _start_workflow(self, wf_name, job_id):
         req.name = wf_name
         req.version = 1
         req.input = {'job_id': job_id}
-        wf_id = self.workflow_client.start_workflow(start_workflow_request=req)
+        wf_id = _retry_on_transient(self.workflow_client.start_workflow,
+                                    start_workflow_request=req)
         logger.info("Started workflow %s: %s", wf_name, wf_id)
         return wf_id
 
@@ -244,11 +274,20 @@ def _wait_for_workflow(self, wf_id, timeout_seconds=90):
         bare status mismatch.
         """
         for _ in range(timeout_seconds):
-            wf = self.workflow_client.get_workflow(wf_id, include_tasks=True)
+            try:
+                wf = self.workflow_client.get_workflow(wf_id, include_tasks=True)
+            except ApiException as e:
+                # A transient blip on a single poll shouldn't abort the wait;
+                # keep polling until the budget is exhausted.
+                if _is_transient(e):
+                    time.sleep(1)
+                    continue
+                raise
             if wf.status in self.TERMINAL_STATES:
                 return wf
             time.sleep(1)
-        wf = self.workflow_client.get_workflow(wf_id, include_tasks=True)
+        wf = _retry_on_transient(self.workflow_client.get_workflow,
+                                 wf_id, include_tasks=True)
         if wf.status not in self.TERMINAL_STATES:
             print(f"  [diag] workflow {wf_id} still {wf.status} "
                   f"after {timeout_seconds}s")

tests/integration/test_lease_extension.py

@@ -36,12 +36,39 @@
 from conductor.client.http.models.task_def import TaskDef
 from conductor.client.http.models.workflow_task import WorkflowTask
 from conductor.client.http.models.start_workflow_request import StartWorkflowRequest
+from conductor.client.http.rest import ApiException
 from conductor.client.orkes.orkes_workflow_client import OrkesWorkflowClient
 from conductor.client.orkes.orkes_metadata_client import OrkesMetadataClient
 
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
+
+def _is_transient(e):
+    """A transient (status 0) ApiException is a raw transport blip against the
+    shared dev server (read timeout, connection reset, HTTP/2 GOAWAY, stale
+    keep-alive) — not a real failure. status 0 means no HTTP response was
+    received at all."""
+    return isinstance(e, ApiException) and (
+        getattr(e, 'transient', False) or e.status in (0, None))
+
+
+def _retry_on_transient(func, *args, retries=4, base_delay=1, **kwargs):
+    """Retry a workflow/metadata client call on a transient (0) transport blip.
+    These `(0)` errors flake the lease suites on a loaded shared server; retrying
+    absorbs the noise. Non-transient errors (real 4xx/5xx) raise immediately."""
+    for attempt in range(retries):
+        try:
+            return func(*args, **kwargs)
+        except ApiException as e:
+            if _is_transient(e) and attempt < retries - 1:
+                logger.warning(
+                    'transient (%s) API error (attempt %d/%d): %s; retrying',
+                    e.status, attempt + 1, retries, e)
+                time.sleep(base_delay * (2 ** attempt))
+                continue
+            raise
+
 # Short response timeout — task must heartbeat to stay alive
 RESPONSE_TIMEOUT_SECONDS = 10
 
@@ -129,9 +156,11 @@ def _register_workflow(self, wf_name, task_name):
         )
         workflow._tasks = [task]
         try:
-            self.metadata_client.update_workflow_def(workflow, overwrite=True)
+            _retry_on_transient(self.metadata_client.update_workflow_def,
+                                 workflow, overwrite=True)
         except Exception:
-            self.metadata_client.register_workflow_def(workflow, overwrite=True)
+            _retry_on_transient(self.metadata_client.register_workflow_def,
+                                 workflow, overwrite=True)
         logger.info("Registered workflow: %s", wf_name)
 
     def _start_workflow(self, wf_name, job_id):
@@ -140,7 +169,8 @@ def _start_workflow(self, wf_name, job_id):
         req.name = wf_name
         req.version = 1
         req.input = {'job_id': job_id}
-        wf_id = self.workflow_client.start_workflow(start_workflow_request=req)
+        wf_id = _retry_on_transient(self.workflow_client.start_workflow,
+                                    start_workflow_request=req)
         logger.info("Started workflow %s: %s", wf_name, wf_id)
         return wf_id
 
@@ -153,11 +183,20 @@ def _wait_for_workflow(self, wf_id, timeout_seconds=60):
         bare status mismatch.
         """
         for i in range(timeout_seconds):
-            wf = self.workflow_client.get_workflow(wf_id, include_tasks=True)
+            try:
+                wf = self.workflow_client.get_workflow(wf_id, include_tasks=True)
+            except ApiException as e:
+                # A transient blip on a single poll shouldn't abort the wait;
+                # keep polling until the budget is exhausted.
+                if _is_transient(e):
+                    time.sleep(1)
+                    continue
+                raise
             if wf.status in self.TERMINAL_STATES:
                 return wf
             time.sleep(1)
-        wf = self.workflow_client.get_workflow(wf_id, include_tasks=True)
+        wf = _retry_on_transient(self.workflow_client.get_workflow,
+                                 wf_id, include_tasks=True)
         if wf.status not in self.TERMINAL_STATES:
             print(f"  [diag] workflow {wf_id} still {wf.status} "
                   f"after {timeout_seconds}s")

tests/integration/test_workflow_client_intg.py

@@ -33,26 +33,47 @@ def get_configuration():
     return configuration
 
 
+def _first_transient_api_exception(exc):
+    """Walk the exception chain (cause/context) and return the first transient
+    ApiException (status 0 / flagged transient), if any.
+
+    Inner test helpers sometimes catch an ApiException and re-raise it as a bare
+    ``Exception`` (losing the type), so we can't rely on the outermost exception
+    type alone. Implicit chaining still records the original on ``__context__``
+    (or ``__cause__`` when ``raise ... from`` is used), so we follow that chain.
+    """
+    seen = set()
+    cur = exc
+    while cur is not None and id(cur) not in seen:
+        seen.add(id(cur))
+        if isinstance(cur, ApiException) and (
+                getattr(cur, 'transient', False) or cur.status in (0, None)):
+            return cur
+        cur = cur.__cause__ or cur.__context__
+    return None
+
+
 def _run_tolerating_transient(label, func, *args, retries=3, **kwargs):
     """Run a sub-suite, retrying only on a transient (status 0) transport blip
     against the shared dev server.
 
-    A `(0)` ApiException is a raw connection/protocol hiccup (stale keep-alive,
-    HTTP/2 GOAWAY race, client closed by a fork cleanup, etc.) — not a real
-    assertion or server failure — and was observed flaking the `test-all`
-    bucket. Retrying absorbs that network noise while still surfacing genuine
-    failures immediately (any non-transient error, or a transient one on the
-    final attempt, re-raises).
+    A `(0)` ApiException is a raw connection/protocol hiccup (read timeout,
+    stale keep-alive, HTTP/2 GOAWAY race, client closed by a fork cleanup, etc.)
+    — not a real assertion or server failure — and was observed flaking the
+    `test-all` bucket. Retrying absorbs that network noise while still surfacing
+    genuine failures immediately (any non-transient error, or a transient one on
+    the final attempt, re-raises). We inspect the whole exception chain because
+    sub-suites may wrap the ApiException in a generic Exception.
     """
     for attempt in range(retries):
         try:
             return func(*args, **kwargs)
-        except ApiException as e:
-            transient = getattr(e, 'transient', False) or e.status in (0, None)
-            if transient and attempt < retries - 1:
+        except Exception as e:
+            transient = _first_transient_api_exception(e)
+            if transient is not None and attempt < retries - 1:
                 logger.warning(
                     'transient (%s) API error in %s (attempt %d/%d): %s; retrying',
-                    e.status, label, attempt + 1, retries, e)
+                    transient.status, label, attempt + 1, retries, transient)
                 time.sleep(2 ** attempt)
                 continue
             raise

tests/integration/workflow/test_workflow_execution.py

@@ -45,6 +45,12 @@ def run_workflow_execution_tests(configuration: Configuration, workflow_executor
         import_modules=['tests.integration.resources.worker.python.python_worker']
     )
     task_handler.start_processes()
+    # Use try/finally (not try/except+re-raise): the sole purpose here is to
+    # stop the workers on the way out. Re-raising as a bare `Exception` used to
+    # discard the original type and traceback, which hid transient
+    # ApiException(status=0) transport blips from the caller's retry logic
+    # (they'd surface as an opaque generic Exception instead). finally cleans up
+    # on both success and failure while letting the original error propagate.
     try:
         scenario_get_workflow_by_correlation_ids(workflow_executor)
         logger.debug('finished workflow correlation ids test')
@@ -64,11 +70,8 @@ def run_workflow_execution_tests(configuration: Configuration, workflow_executor
         logger.debug('finished execute_workflow error handling tests')
         run_signal_tests(configuration, workflow_executor)
         logger.debug('finished signal API tests')
-
-    except Exception as e:
+    finally:
         task_handler.stop_processes()
-        raise Exception(f'failed integration tests, reason: {e}')
-    task_handler.stop_processes()
 
 
 def generate_tasks_defs():