Skip to content

Commit d71af0f

Browse files
authored
Add continue-as-new version support (#254)
* Add continue-as-new version support Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: df26610a-beb4-44ea-a04a-6aedeb316d2a * Use equality in version presence assertion Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: df26610a-beb4-44ea-a04a-6aedeb316d2a --------- Copilot-Session: df26610a-beb4-44ea-a04a-6aedeb316d2a
1 parent dc477ef commit d71af0f

5 files changed

Lines changed: 28 additions & 8 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,12 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
77

88
## Unreleased
99

10+
ADDED
11+
12+
- Added the optional `new_version` argument to
13+
`OrchestrationContext.continue_as_new()` so continued orchestrations can
14+
switch to a new version.
15+
1016
## v1.9.0
1117

1218
ADDED

‎durabletask/internal/helpers.py‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -300,12 +300,14 @@ def new_complete_orchestration_action(
300300
status: pb.OrchestrationStatus,
301301
result: str | None = None,
302302
failure_details: pb.TaskFailureDetails | None = None,
303-
carryover_events: list[pb.HistoryEvent] | None = None) -> pb.OrchestratorAction:
303+
carryover_events: list[pb.HistoryEvent] | None = None,
304+
new_version: str | None = None) -> pb.OrchestratorAction:
304305
completeOrchestrationAction = pb.CompleteOrchestrationAction(
305306
orchestrationStatus=status,
306307
result=get_string_value(result),
307308
failureDetails=failure_details,
308-
carryoverEvents=carryover_events)
309+
carryoverEvents=carryover_events,
310+
newVersion=get_string_value(new_version))
309311

310312
return pb.OrchestratorAction(id=id, completeOrchestration=completeOrchestrationAction)
311313

‎durabletask/task.py‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -389,7 +389,8 @@ def send_event(self, instance_id: str, event_name: str, *,
389389
)
390390

391391
@abstractmethod
392-
def continue_as_new(self, new_input: Any, *, save_events: bool = False) -> None:
392+
def continue_as_new(self, new_input: Any, *, save_events: bool = False,
393+
new_version: str | None = None) -> None:
393394
"""Continue the orchestration execution as a new instance.
394395
395396
Parameters
@@ -398,6 +399,8 @@ def continue_as_new(self, new_input: Any, *, save_events: bool = False) -> None:
398399
The new input to use for the new orchestration instance.
399400
save_events : bool
400401
A flag indicating whether to add any unprocessed external events in the new orchestration history.
402+
new_version : str | None
403+
An optional version to assign to the new orchestration instance.
401404
"""
402405
pass
403406

‎durabletask/worker.py‎

Lines changed: 8 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1523,6 +1523,7 @@ def __init__(self,
15231523
self._received_events: dict[str, list[str | None]] = {}
15241524
self._pending_events: dict[str, list[task.CancellableTask[Any]]] = {}
15251525
self._new_input: Any | None = None
1526+
self._new_version: str | None = None
15261527
self._save_events = False
15271528
self._encoded_custom_status: str | None = None
15281529
self._parent_trace_context: pb.TraceContext | None = None
@@ -1619,7 +1620,8 @@ def set_failed(self, ex: Exception | pb.TaskFailureDetails):
16191620
)
16201621
self._pending_actions[action.id] = action
16211622

1622-
def set_continued_as_new(self, new_input: Any, save_events: bool):
1623+
def set_continued_as_new(self, new_input: Any, save_events: bool,
1624+
new_version: str | None = None):
16231625
if self._is_complete:
16241626
return
16251627

@@ -1633,6 +1635,7 @@ def set_continued_as_new(self, new_input: Any, save_events: bool):
16331635
# self._pending_actions.clear() # Cancel any pending actions
16341636
self._completion_status = pb.ORCHESTRATION_STATUS_CONTINUED_AS_NEW
16351637
self._new_input = new_input
1638+
self._new_version = new_version
16361639
self._save_events = save_events
16371640

16381641
def get_actions(self) -> list[pb.OrchestratorAction]:
@@ -1657,6 +1660,7 @@ def get_actions(self) -> list[pb.OrchestratorAction]:
16571660
result=self._data_converter.serialize(self._new_input),
16581661
failure_details=None,
16591662
carryover_events=carryover_events,
1663+
new_version=self._new_version,
16601664
)
16611665
# We must return the existing tasks as well, to capture entity unlocks
16621666
current_actions.append(action)
@@ -2135,11 +2139,12 @@ def send_event(self, instance_id: str, event_name: str, *,
21352139
),
21362140
)
21372141

2138-
def continue_as_new(self, new_input: Any, *, save_events: bool = False) -> None:
2142+
def continue_as_new(self, new_input: Any, *, save_events: bool = False,
2143+
new_version: str | None = None) -> None:
21392144
if self._is_complete:
21402145
return
21412146

2142-
self.set_continued_as_new(new_input, save_events)
2147+
self.set_continued_as_new(new_input, save_events, new_version)
21432148

21442149
def new_uuid(self) -> str:
21452150
NAMESPACE_UUID: str = "9e952958-5e33-4daf-827f-2fa12937b875"

‎tests/durabletask/test_orchestration_executor.py‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1719,11 +1719,12 @@ def orchestrator(ctx: task.OrchestrationContext, _):
17191719

17201720

17211721
@pytest.mark.parametrize("save_events", [True, False])
1722-
def test_continue_as_new(save_events: bool):
1722+
@pytest.mark.parametrize("new_version", [None, "v2"])
1723+
def test_continue_as_new(save_events: bool, new_version: str | None):
17231724
"""Tests the behavior of the continue-as-new API"""
17241725
def orchestrator(ctx: task.OrchestrationContext, input: int):
17251726
yield ctx.create_timer(ctx.current_utc_datetime + timedelta(days=1))
1726-
ctx.continue_as_new(input + 1, save_events=save_events)
1727+
ctx.continue_as_new(input + 1, save_events=save_events, new_version=new_version)
17271728

17281729
registry = worker._Registry()
17291730
orchestrator_name = registry.add_orchestrator(orchestrator)
@@ -1745,6 +1746,9 @@ def orchestrator(ctx: task.OrchestrationContext, input: int):
17451746
complete_action = get_and_validate_complete_orchestration_action_list(1, actions)
17461747
assert complete_action.orchestrationStatus == pb.ORCHESTRATION_STATUS_CONTINUED_AS_NEW
17471748
assert complete_action.result.value == json.dumps(2)
1749+
assert complete_action.HasField("newVersion") == (new_version is not None)
1750+
if new_version is not None:
1751+
assert complete_action.newVersion.value == new_version
17481752
assert len(complete_action.carryoverEvents) == (3 if save_events else 0)
17491753
for i in range(len(complete_action.carryoverEvents)):
17501754
event = complete_action.carryoverEvents[i]

0 commit comments

Comments
 (0)