CH-302 feat: Add optional async workflow submission timeout

docs/applications/development/workflows-api.md

@@ -57,6 +57,25 @@ op = operations.ParallelOperation('test-parallel-op-', (tasks.PythonTask('p1', f
     - **ParallelOperation** - takes a list of tasks running in parallel
     - **SimpleDagOperation** - simple Direct-Acyclic-Graph made from pipeline of parallel tasks
 
+### Asynchronous submission timeout
+
+Asynchronous operations such as `PipelineOperation` and `ParallelOperation` accept
+an optional HTTP submission timeout:
+
+```python
+workflow = op.execute(request_timeout=(2, 10))  # connection and read timeouts, in seconds
+```
+
+A single number is also accepted. Omitting `request_timeout` preserves the existing
+submission behaviour. `ContainerizedOperation.submit()` and
+`argo_service.submit_workflow()` accept the same keyword.
+
+This bounds the submission request, not the workflow's execution time. A timeout
+does not prove that Argo rejected the workflow; callers must reconcile uncertain
+outcomes before retrying. CloudHarness does not automatically resubmit.
+The existing `timeout` argument on synchronous operations still controls waiting
+for completion and is unchanged.
+
 ### Task Types
 - _InlinedTask_ - runs a script in a default image. No specific container configuration
   - **PythonTask** - runs a Python function
@@ -341,4 +360,4 @@ Workflows can be monitored through argo ui going to argo.[DOMAIN] or through com
 
 ## More examples
 See the [samples application controller](../../../applications/samples/backend/samples/controllers/workflows_controller.py) for a practical case of a service using asynchronous and synchronous workflows as part of the api.
-Some examples are also available as unit tests.
+Some examples are also available as unit tests.

libraries/cloudharness-common/cloudharness/workflows/argo_service.py

@@ -225,15 +225,16 @@ def get_workflows(status=None, limit=10, continue_token=None, timeout_seconds=3,
         return SearchResult(api_response)
 
 
-def submit_workflow(spec) -> Workflow:
+def submit_workflow(spec, *, request_timeout=None) -> Workflow:
+    """Create one workflow; optionally bound HTTP I/O with seconds or (connect, read)."""
     log.debug(f"Submitting workflow %s", spec)
 
-    service = WorkflowServiceApi(api_client=get_api_client())
-
     req = V1alpha1WorkflowCreateRequest(workflow=spec, instance_id=namespace, namespace=namespace, _check_type=False)
 
-    # pprint(service.list_workflows('ch', V1alpha1WorkflowList()))
-    wf = service.create_workflow(namespace, req)
+    options = {} if request_timeout is None else {"_request_timeout": request_timeout}
+    with get_api_client() as api_client:
+        service = WorkflowServiceApi(api_client=api_client)
+        wf = service.create_workflow(namespace, req, **options)
     return Workflow(wf)
 
 

libraries/cloudharness-common/cloudharness/workflows/operations.py

@@ -211,14 +211,14 @@ def modify_template(self, template):
 
         return template
 
-    def submit(self):
-        """Created and submits the Argo workflow"""
+    def submit(self, **kwargs):
+        """Submit the workflow, forwarding keyword arguments to argo_service.submit_workflow."""
         op = self.to_workflow()
 
         log.debug("Submitting workflow\n" + yaml.dump(op))
 
         # TODO use rest api for that? Include this into cloudharness.workflows?
-        self.persisted = argo_service.submit_workflow(op)
+        self.persisted = argo_service.submit_workflow(op, **kwargs)
         return self.persisted
 
     def is_running(self):
@@ -347,9 +347,9 @@ class AsyncOperation(ContainerizedOperation):
     """The operation is made asynchronously in an Argo workflow.
     The workflow can be monitored during the execution"""
 
-    def execute(self):
-        op = self.submit()
-        return op
+    def execute(self, **kwargs):
+        """Submit asynchronously, forwarding keyword arguments to submit."""
+        return self.submit(**kwargs)
 
     def get_operation_update_url(self):
         return f"{env.get_cloudharness_workflows_service_url()}/api/operations/{self.persisted.name}"

libraries/cloudharness-common/tests/test_argo_client.py

@@ -60,6 +60,50 @@ def test_submit_preserves_raw_workflow_and_future_fields(api, workflow):
     assert transport.call_args.args[0] == "POST"
 
 
+@pytest.mark.parametrize("operation_type", [operations.PipelineOperation, operations.ParallelOperation])
+@pytest.mark.parametrize("request_timeout", [None, 1.5, (2, 10)])
+def test_async_execute_forwards_optional_submission_timeout(api, workflow, operation_type, request_timeout):
+    respond, client = api
+    transport = respond(workflow)
+    close = Mock(wraps=client.close)
+    client.close = close
+    operation = operation_type("sdk-timeout-", [])
+
+    result = operation.execute(request_timeout=request_timeout)
+
+    assert result.name == "sdk-test"
+    assert result.raw.status.phase == "Running"
+    assert operation.persisted is result
+    assert transport.call_count == 1
+    assert transport.call_args.kwargs["_request_timeout"] == request_timeout
+    close.assert_called_once_with()
+
+
+def test_async_execute_default_preserves_no_argument_submit_override():
+    operation = operations.PipelineOperation("legacy-submit-", [])
+    expected = object()
+    operation.submit = lambda: expected
+
+    assert operation.execute() is expected
+
+
+def test_submit_closes_client_and_preserves_errors_without_retry(api, monkeypatch):
+    _, client = api
+    close = Mock(wraps=client.close)
+    monkeypatch.setattr(client, "close", close)
+    error = exceptions.ApiException(status=503, reason="Unavailable")
+    error.body = b"unavailable"
+    transport = Mock(side_effect=error)
+    monkeypatch.setattr(client.rest_client, "request", transport)
+
+    with pytest.raises(exceptions.ApiException) as caught:
+        argo_service.submit_workflow({"metadata": {"name": "sdk-test"}}, request_timeout=(2, 10))
+
+    assert caught.value is error
+    assert transport.call_count == 1
+    close.assert_called_once_with()
+
+
 def test_running_response_keeps_native_progress_and_optional_fields(api, workflow):
     respond, _ = api
     respond(workflow)