Expose continue-as-new backoff start interval (#1613)

* Expose continue-as-new backoff start interval

* Add changelog note

Author: Sushisource

CHANGELOG.md

@@ -20,6 +20,8 @@ to include examples, links to docs, or any other relevant information.
 
 ### Added
 
+- Exposed `backoff_start_interval` for continue-as-new, to allow the new workflow to start after a delay.
+
 ### Changed
 
 - AWS Lambda worker `configure` parameter supports sync, async, and async

temporalio/bridge/Cargo.lock

@@ -942,6 +942,7 @@ class ContinueAsNewWorkflowExecution(google.protobuf.message.Message):
     RETRY_POLICY_FIELD_NUMBER: builtins.int
     VERSIONING_INTENT_FIELD_NUMBER: builtins.int
     INITIAL_VERSIONING_BEHAVIOR_FIELD_NUMBER: builtins.int
+    BACKOFF_START_INTERVAL_FIELD_NUMBER: builtins.int
     workflow_type: builtins.str
     """The identifier the lang-specific sdk uses to execute workflow code"""
     task_queue: builtins.str
@@ -1000,6 +1001,9 @@ class ContinueAsNewWorkflowExecution(google.protobuf.message.Message):
     For example, choose to AutoUpgrade on continue-as-new instead of inheriting the pinned version
     of the previous run.
     """
+    @property
+    def backoff_start_interval(self) -> google.protobuf.duration_pb2.Duration:
+        """Delay before the first workflow task of the continued run is scheduled."""
     def __init__(
         self,
         *,
@@ -1024,10 +1028,13 @@ class ContinueAsNewWorkflowExecution(google.protobuf.message.Message):
         retry_policy: temporalio.api.common.v1.message_pb2.RetryPolicy | None = ...,
         versioning_intent: temporalio.bridge.proto.common.common_pb2.VersioningIntent.ValueType = ...,
         initial_versioning_behavior: temporalio.api.enums.v1.workflow_pb2.ContinueAsNewVersioningBehavior.ValueType = ...,
+        backoff_start_interval: google.protobuf.duration_pb2.Duration | None = ...,
     ) -> None: ...
     def HasField(
         self,
         field_name: typing_extensions.Literal[
+            "backoff_start_interval",
+            b"backoff_start_interval",
             "retry_policy",
             b"retry_policy",
             "search_attributes",
@@ -1043,6 +1050,8 @@ class ContinueAsNewWorkflowExecution(google.protobuf.message.Message):
         field_name: typing_extensions.Literal[
             "arguments",
             b"arguments",
+            "backoff_start_interval",
+            b"backoff_start_interval",
             "headers",
             b"headers",
             "initial_versioning_behavior",

temporalio/bridge/proto/workflow_commands/workflow_commands_pb2.py

@@ -165,6 +165,7 @@ class ContinueAsNewInput:
     task_queue: str | None
     run_timeout: timedelta | None
     task_timeout: timedelta | None
+    backoff_start_interval: timedelta | None
     retry_policy: temporalio.common.RetryPolicy | None
     memo: Mapping[str, Any] | None
     search_attributes: None | (

temporalio/bridge/proto/workflow_commands/workflow_commands_pb2.pyi

@@ -1148,6 +1148,7 @@ def workflow_continue_as_new(
         task_queue: str | None,
         run_timeout: timedelta | None,
         task_timeout: timedelta | None,
+        backoff_start_interval: timedelta | None,
         retry_policy: temporalio.common.RetryPolicy | None,
         memo: Mapping[str, Any] | None,
         search_attributes: None
@@ -1178,6 +1179,7 @@ def workflow_continue_as_new(
                 task_queue=task_queue,
                 run_timeout=run_timeout,
                 task_timeout=task_timeout,
+                backoff_start_interval=backoff_start_interval,
                 retry_policy=retry_policy,
                 memo=memo,
                 search_attributes=search_attributes,
@@ -3587,6 +3589,8 @@ def _apply_command(self) -> None:
             v.workflow_run_timeout.FromTimedelta(self._input.run_timeout)
         if self._input.task_timeout:
             v.workflow_task_timeout.FromTimedelta(self._input.task_timeout)
+        if self._input.backoff_start_interval:
+            v.backoff_start_interval.FromTimedelta(self._input.backoff_start_interval)
         if self._input.headers:
             temporalio.common._apply_headers(self._input.headers, v.headers)
         if self._input.retry_policy:

temporalio/bridge/sdk-core

@@ -286,6 +286,7 @@ def workflow_continue_as_new(
         task_queue: str | None,
         run_timeout: timedelta | None,
         task_timeout: timedelta | None,
+        backoff_start_interval: timedelta | None,
         retry_policy: temporalio.common.RetryPolicy | None,
         memo: Mapping[str, Any] | None,
         search_attributes: None

temporalio/worker/_interceptor.py

@@ -683,6 +683,7 @@ def continue_as_new(
     task_queue: str | None = None,
     run_timeout: timedelta | None = None,
     task_timeout: timedelta | None = None,
+    backoff_start_interval: timedelta | None = None,
     retry_policy: temporalio.common.RetryPolicy | None = None,
     memo: Mapping[str, Any] | None = None,
     search_attributes: None
@@ -702,6 +703,7 @@ def continue_as_new(
     task_queue: str | None = None,
     run_timeout: timedelta | None = None,
     task_timeout: timedelta | None = None,
+    backoff_start_interval: timedelta | None = None,
     retry_policy: temporalio.common.RetryPolicy | None = None,
     memo: Mapping[str, Any] | None = None,
     search_attributes: None
@@ -722,6 +724,7 @@ def continue_as_new(
     task_queue: str | None = None,
     run_timeout: timedelta | None = None,
     task_timeout: timedelta | None = None,
+    backoff_start_interval: timedelta | None = None,
     retry_policy: temporalio.common.RetryPolicy | None = None,
     memo: Mapping[str, Any] | None = None,
     search_attributes: None
@@ -742,6 +745,7 @@ def continue_as_new(
     task_queue: str | None = None,
     run_timeout: timedelta | None = None,
     task_timeout: timedelta | None = None,
+    backoff_start_interval: timedelta | None = None,
     retry_policy: temporalio.common.RetryPolicy | None = None,
     memo: Mapping[str, Any] | None = None,
     search_attributes: None
@@ -762,6 +766,7 @@ def continue_as_new(
     task_queue: str | None = None,
     run_timeout: timedelta | None = None,
     task_timeout: timedelta | None = None,
+    backoff_start_interval: timedelta | None = None,
     retry_policy: temporalio.common.RetryPolicy | None = None,
     memo: Mapping[str, Any] | None = None,
     search_attributes: None
@@ -781,6 +786,7 @@ def continue_as_new(
     task_queue: str | None = None,
     run_timeout: timedelta | None = None,
     task_timeout: timedelta | None = None,
+    backoff_start_interval: timedelta | None = None,
     retry_policy: temporalio.common.RetryPolicy | None = None,
     memo: Mapping[str, Any] | None = None,
     search_attributes: None
@@ -804,6 +810,8 @@ def continue_as_new(
             workflow's run timeout.
         task_timeout: Timeout of a single workflow task. Defaults to the current
             workflow's task timeout.
+        backoff_start_interval: Delay before the first workflow task of the
+            continued run is scheduled.
         memo: Memo for the workflow. Defaults to the current workflow's memo.
         search_attributes: Search attributes for the workflow. Defaults to the
             current workflow's search attributes. The dictionary form of this is
@@ -826,6 +834,7 @@ def continue_as_new(
         task_queue=task_queue,
         run_timeout=run_timeout,
         task_timeout=task_timeout,
+        backoff_start_interval=backoff_start_interval,
         retry_policy=retry_policy,
         memo=memo,
         search_attributes=search_attributes,

temporalio/worker/_workflow_instance.py

@@ -1893,6 +1893,7 @@ async def run(self, past_run_ids: list[str]) -> list[str]:
             # Add memo and retry policy to check
             memo={"past_run_id_count": len(past_run_ids)},
             retry_policy=RetryPolicy(maximum_attempts=1000 + len(past_run_ids)),
+            backoff_start_interval=timedelta(milliseconds=1),
         )
 
 
@@ -1914,6 +1915,19 @@ async def test_workflow_continue_as_new(client: Client, env: WorkflowEnvironment
         result = await handle.result()
         assert len(result) == 5
         assert result[0] == handle.first_execution_run_id
+        first_run_handle = client.get_workflow_handle(
+            handle.id, run_id=handle.first_execution_run_id
+        )
+        history = await first_run_handle.fetch_history()
+        continued = [
+            event.workflow_execution_continued_as_new_event_attributes
+            for event in history.events
+            if event.HasField("workflow_execution_continued_as_new_event_attributes")
+        ]
+        assert continued
+        assert continued[0].backoff_start_interval.ToTimedelta() == timedelta(
+            milliseconds=1
+        )
 
 
 sa_prefix = "python_test_"