Skip to content

Commit 0072642

Browse files
committed
Detect send event target changes during replay
Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 56111c85-99ce-477c-a97f-c2d57a162a55
1 parent ed7be01 commit 0072642

4 files changed

Lines changed: 45 additions & 2 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,9 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
1010
ADDED
1111

1212
- Added `OrchestrationContext.send_event()` for replay-safe, one-way event
13-
delivery from an orchestration to another orchestration instance.
13+
delivery from an orchestration to another orchestration instance. During
14+
replay, Python validates the target instance ID as well as the event name,
15+
intentionally detecting target changes that DurableTask.Core does not.
1416
- Added asynchronous `AsyncScheduledTaskClient` / `AsyncScheduleClient` and
1517
`AsyncExportHistoryClient` / `AsyncExportHistoryJobClient` APIs. Applications
1618
using `AsyncTaskHubGrpcClient` can now manage scheduled tasks and history

‎durabletask/task.py‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -370,6 +370,9 @@ def send_event(self, instance_id: str, event_name: str, *,
370370
:meth:`wait_for_external_event`. This is a one-way operation and does
371371
not wait for the target orchestration to process the event. If the
372372
target orchestration does not exist, the event is silently dropped.
373+
During replay, the Python SDK validates both the event name and target
374+
instance ID. This is intentionally stricter than DurableTask.Core,
375+
which does not validate the target instance ID.
373376
374377
Parameters
375378
----------

‎durabletask/worker.py‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3017,6 +3017,19 @@ def _cancel_timer() -> None:
30173017
task.get_name(ctx.send_event),
30183018
action,
30193019
)
3020+
elif (
3021+
action.sendEvent.instance.instanceId
3022+
!= event.eventSent.instanceId
3023+
):
3024+
# Python intentionally validates the target during replay,
3025+
# unlike DurableTask.Core, to avoid silently suppressing
3026+
# delivery when orchestration code changes the target.
3027+
raise _get_wrong_action_target_error(
3028+
event.eventId,
3029+
method_name=task.get_name(ctx.send_event),
3030+
expected_instance_id=event.eventSent.instanceId,
3031+
actual_instance_id=action.sendEvent.instance.instanceId,
3032+
)
30203033
elif (
30213034
action.sendEvent.name.casefold()
30223035
!= event.eventSent.name.casefold()
@@ -3340,6 +3353,21 @@ def _get_wrong_action_name_error(
33403353
)
33413354

33423355

3356+
def _get_wrong_action_target_error(
3357+
task_id: int,
3358+
method_name: str,
3359+
expected_instance_id: str,
3360+
actual_instance_id: str,
3361+
) -> task.NonDeterminismError:
3362+
return task.NonDeterminismError(
3363+
f"Failed to restore orchestration state due to a history mismatch: A previous execution called "
3364+
f"{method_name} with target instance ID='{expected_instance_id}' and sequence number {task_id}, but the "
3365+
f"current execution is instead targeting instance ID='{actual_instance_id}' as part of rebuilding its "
3366+
f"history. This kind of mismatch can happen if an orchestration has non-deterministic logic or if the "
3367+
f"code was changed after an instance of this orchestration already started running."
3368+
)
3369+
3370+
33433371
def _get_method_name_for_action(action: pb.OrchestratorAction) -> str:
33443372
action_type = action.WhichOneof("orchestratorActionType")
33453373
match action_type:

‎tests/durabletask/test_orchestration_executor.py‎

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1558,7 +1558,10 @@ def orchestrator(ctx: task.OrchestrationContext, _):
15581558
assert result.actions == []
15591559

15601560

1561-
@pytest.mark.parametrize("mismatch", ["name", "type", "missing", "entity"])
1561+
@pytest.mark.parametrize(
1562+
"mismatch",
1563+
["name", "target", "type", "missing", "entity"],
1564+
)
15621565
def test_send_event_replay_detects_nondeterminism(mismatch: str):
15631566
if mismatch == "type":
15641567
def orchestrator(ctx: task.OrchestrationContext, _):
@@ -1574,6 +1577,10 @@ def orchestrator(ctx: task.OrchestrationContext, _):
15741577
"increment",
15751578
)
15761579
yield ctx.wait_for_external_event("Done")
1580+
elif mismatch == "target":
1581+
def orchestrator(ctx: task.OrchestrationContext, _):
1582+
ctx.send_event("changed-target", "OriginalName")
1583+
yield ctx.wait_for_external_event("Done")
15771584
else:
15781585
def orchestrator(ctx: task.OrchestrationContext, _):
15791586
ctx.send_event("target-instance", "ChangedName")
@@ -1605,6 +1612,9 @@ def orchestrator(ctx: task.OrchestrationContext, _):
16051612
if mismatch == "name":
16061613
assert "OriginalName" in complete_action.failureDetails.errorMessage
16071614
assert "ChangedName" in complete_action.failureDetails.errorMessage
1615+
elif mismatch == "target":
1616+
assert "target-instance" in complete_action.failureDetails.errorMessage
1617+
assert "changed-target" in complete_action.failureDetails.errorMessage
16081618
elif mismatch == "type":
16091619
assert "create_timer" in complete_action.failureDetails.errorMessage
16101620
elif mismatch == "entity":

0 commit comments

Comments
 (0)