Skip to content

Populate timer origin field #976

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
acroca wants to merge 14 commits into
base: main
Choose a base branch
from
Open

Populate timer origin field #976

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
fae65e9
Populate timer origin field
acroca Apr 10, 2026
b6d99bd
ruff
acroca Apr 10, 2026
b2c7d87
Backwards compatible external event timer
acroca Apr 13, 2026
9485df6
ruff
acroca Apr 13, 2026
03558b8
Addressed review comments
acroca Apr 13, 2026
6c4e337
Fix ci
acroca Apr 13, 2026
ad30808
Addressed review comments
acroca Apr 13, 2026
51222cf
Addressed review comments
acroca Apr 13, 2026
461620e
Fix CI
acroca Apr 13, 2026
5a79404
Remove pending external_event_task when timeout fires
acroca Apr 14, 2026
7805f97
Addressed review comments
acroca Apr 14, 2026
eae6655
Addressed review comments
acroca Apr 15, 2026
c877ddd
Merge remote-tracking branch 'upstream/main' into timers-origin
acroca Apr 15, 2026
bd9297a
Lint
acroca Apr 15, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions dapr/clients/grpc/_helpers.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,9 @@ def unpack(data: GrpcAny, message: GrpcMessage) -> None:
"""
if not isinstance(message, GrpcMessage):
raise ValueError('output message is not protocol buffer message object')
# cast: newer types-grpcio stubs declare ``message.DESCRIPTOR`` as a union of
# ``Descriptor`` and the C-extension ``_upb._message.Descriptor``. Both are
# accepted at runtime by ``Any.Is``; the cast narrows the type for mypy.
if not data.Is(cast(Descriptor, message.DESCRIPTOR)):
raise ValueError(f'invalid type. serialized message type: {data.type_url}')
data.Unpack(message)
Expand Down
35 changes: 9 additions & 26 deletions ext/dapr-ext-workflow/dapr/ext/workflow/_durabletask/internal/helpers.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,15 @@
from typing import Optional

import dapr.ext.workflow._durabletask.internal.protos as pb
from dapr.ext.workflow._durabletask.internal.timer import ( # noqa: F401
OPTIONAL_TIMER_FIRE_AT,
TimerOrigin,
is_optional_timer_action,
is_optional_timer_event,
new_create_timer_action,
new_timer_created_event,
new_timer_fired_event,
)
from google.protobuf import timestamp_pb2, wrappers_pb2

# TODO: The new_xxx_event methods are only used by test code and should be moved elsewhere
Expand All @@ -40,26 +49,6 @@ def new_execution_started_event(
)


def new_timer_created_event(timer_id: int, fire_at: datetime) -> pb.HistoryEvent:
ts = timestamp_pb2.Timestamp()
ts.FromDatetime(fire_at)
return pb.HistoryEvent(
eventId=timer_id,
timestamp=timestamp_pb2.Timestamp(),
timerCreated=pb.TimerCreatedEvent(fireAt=ts),
)


def new_timer_fired_event(timer_id: int, fire_at: datetime) -> pb.HistoryEvent:
ts = timestamp_pb2.Timestamp()
ts.FromDatetime(fire_at)
return pb.HistoryEvent(
eventId=-1,
timestamp=timestamp_pb2.Timestamp(),
timerFired=pb.TimerFiredEvent(fireAt=ts, timerId=timer_id),
)


def new_task_scheduled_event(
event_id: int, name: str, encoded_input: Optional[str] = None
) -> pb.HistoryEvent:
Expand Down Expand Up @@ -202,12 +191,6 @@ def new_workflow_version_not_available_action(
)


def new_create_timer_action(id: int, fire_at: datetime) -> pb.WorkflowAction:
timestamp = timestamp_pb2.Timestamp()
timestamp.FromDatetime(fire_at)
return pb.WorkflowAction(id=id, createTimer=pb.CreateTimerAction(fireAt=timestamp))


def new_schedule_task_action(
id: int,
name: str,
Expand Down
50 changes: 27 additions & 23 deletions ext/dapr-ext-workflow/dapr/ext/workflow/_durabletask/internal/history_events_pb2.py
View file
Open in desktop

Large diffs are not rendered by default.

52 changes: 49 additions & 3 deletions ext/dapr-ext-workflow/dapr/ext/workflow/_durabletask/internal/history_events_pb2.pyi
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -351,6 +351,44 @@ class TimerOriginExternalEvent(_message.Message):

Global___TimerOriginExternalEvent: _TypeAlias = TimerOriginExternalEvent # noqa: Y015

@_typing.final
class TimerOriginActivityRetry(_message.Message):
"""Indicates the timer was created as a retry delay for an activity execution."""

DESCRIPTOR: _descriptor.Descriptor

TASKEXECUTIONID_FIELD_NUMBER: _builtins.int
Copy link
Copy Markdown
Contributor

@sicoyle sicoyle Apr 14, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
TASKEXECUTIONID_FIELD_NUMBER: _builtins.int
TASK_EXECUTION_ID_FIELD_NUMBER: _builtins.int

why do we need to track the field number for this? can it just be TASK_EXECUTION_ID instead?

Copy link
Copy Markdown
Member Author

@acroca acroca Apr 15, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is generated from the protos: https://github.com/dapr/durabletask-protobuf/blob/3380843f88b6df933600d16502725fe99591bdee/protos/history_events.proto#L96

taskExecutionId: _builtins.str
"""The task execution ID of the activity being retried."""
def __init__(
self,
*,
taskExecutionId: _builtins.str = ...,
) -> None: ...
_ClearFieldArgType: _TypeAlias = _typing.Literal["taskExecutionId", b"taskExecutionId"] # noqa: Y015
def ClearField(self, field_name: _ClearFieldArgType) -> None: ...

Global___TimerOriginActivityRetry: _TypeAlias = TimerOriginActivityRetry # noqa: Y015

@_typing.final
class TimerOriginChildWorkflowRetry(_message.Message):
"""Indicates the timer was created as a retry delay for a child workflow execution."""

DESCRIPTOR: _descriptor.Descriptor

INSTANCEID_FIELD_NUMBER: _builtins.int
instanceId: _builtins.str
"""The instance ID of the workflow being retried."""
def __init__(
self,
*,
instanceId: _builtins.str = ...,
) -> None: ...
_ClearFieldArgType: _TypeAlias = _typing.Literal["instanceId", b"instanceId"] # noqa: Y015
def ClearField(self, field_name: _ClearFieldArgType) -> None: ...

Global___TimerOriginChildWorkflowRetry: _TypeAlias = TimerOriginChildWorkflowRetry # noqa: Y015

@_typing.final
class TimerCreatedEvent(_message.Message):
DESCRIPTOR: _descriptor.Descriptor
Expand All @@ -360,6 +398,8 @@ class TimerCreatedEvent(_message.Message):
RERUNPARENTINSTANCEINFO_FIELD_NUMBER: _builtins.int
CREATETIMER_FIELD_NUMBER: _builtins.int
EXTERNALEVENT_FIELD_NUMBER: _builtins.int
ACTIVITYRETRY_FIELD_NUMBER: _builtins.int
CHILDWORKFLOWRETRY_FIELD_NUMBER: _builtins.int
name: _builtins.str
@_builtins.property
def fireAt(self) -> _timestamp_pb2.Timestamp: ...
Expand All @@ -373,6 +413,10 @@ class TimerCreatedEvent(_message.Message):
def createTimer(self) -> Global___TimerOriginCreateTimer: ...
@_builtins.property
def externalEvent(self) -> Global___TimerOriginExternalEvent: ...
@_builtins.property
def activityRetry(self) -> Global___TimerOriginActivityRetry: ...
@_builtins.property
def childWorkflowRetry(self) -> Global___TimerOriginChildWorkflowRetry: ...
def __init__(
self,
*,
Expand All @@ -381,16 +425,18 @@ class TimerCreatedEvent(_message.Message):
rerunParentInstanceInfo: _orchestration_pb2.RerunParentInstanceInfo | None = ...,
createTimer: Global___TimerOriginCreateTimer | None = ...,
externalEvent: Global___TimerOriginExternalEvent | None = ...,
activityRetry: Global___TimerOriginActivityRetry | None = ...,
childWorkflowRetry: Global___TimerOriginChildWorkflowRetry | None = ...,
) -> None: ...
_HasFieldArgType: _TypeAlias = _typing.Literal["_name", b"_name", "_rerunParentInstanceInfo", b"_rerunParentInstanceInfo", "createTimer", b"createTimer", "externalEvent", b"externalEvent", "fireAt", b"fireAt", "name", b"name", "origin", b"origin", "rerunParentInstanceInfo", b"rerunParentInstanceInfo"] # noqa: Y015
_HasFieldArgType: _TypeAlias = _typing.Literal["_name", b"_name", "_rerunParentInstanceInfo", b"_rerunParentInstanceInfo", "activityRetry", b"activityRetry", "childWorkflowRetry", b"childWorkflowRetry", "createTimer", b"createTimer", "externalEvent", b"externalEvent", "fireAt", b"fireAt", "name", b"name", "origin", b"origin", "rerunParentInstanceInfo", b"rerunParentInstanceInfo"] # noqa: Y015
def HasField(self, field_name: _HasFieldArgType) -> _builtins.bool: ...
_ClearFieldArgType: _TypeAlias = _typing.Literal["_name", b"_name", "_rerunParentInstanceInfo", b"_rerunParentInstanceInfo", "createTimer", b"createTimer", "externalEvent", b"externalEvent", "fireAt", b"fireAt", "name", b"name", "origin", b"origin", "rerunParentInstanceInfo", b"rerunParentInstanceInfo"] # noqa: Y015
_ClearFieldArgType: _TypeAlias = _typing.Literal["_name", b"_name", "_rerunParentInstanceInfo", b"_rerunParentInstanceInfo", "activityRetry", b"activityRetry", "childWorkflowRetry", b"childWorkflowRetry", "createTimer", b"createTimer", "externalEvent", b"externalEvent", "fireAt", b"fireAt", "name", b"name", "origin", b"origin", "rerunParentInstanceInfo", b"rerunParentInstanceInfo"] # noqa: Y015
def ClearField(self, field_name: _ClearFieldArgType) -> None: ...
_WhichOneofReturnType__name: _TypeAlias = _typing.Literal["name"] # noqa: Y015
_WhichOneofArgType__name: _TypeAlias = _typing.Literal["_name", b"_name"] # noqa: Y015
_WhichOneofReturnType__rerunParentInstanceInfo: _TypeAlias = _typing.Literal["rerunParentInstanceInfo"] # noqa: Y015
_WhichOneofArgType__rerunParentInstanceInfo: _TypeAlias = _typing.Literal["_rerunParentInstanceInfo", b"_rerunParentInstanceInfo"] # noqa: Y015
_WhichOneofReturnType_origin: _TypeAlias = _typing.Literal["createTimer", "externalEvent"] # noqa: Y015
_WhichOneofReturnType_origin: _TypeAlias = _typing.Literal["createTimer", "externalEvent", "activityRetry", "childWorkflowRetry"] # noqa: Y015
_WhichOneofArgType_origin: _TypeAlias = _typing.Literal["origin", b"origin"] # noqa: Y015
@_typing.overload
def WhichOneof(self, oneof_group: _WhichOneofArgType__name) -> _WhichOneofReturnType__name | None: ...
Expand Down
24 changes: 12 additions & 12 deletions ext/dapr-ext-workflow/dapr/ext/workflow/_durabletask/internal/orchestrator_actions_pb2.py
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

14 changes: 11 additions & 3 deletions ext/dapr-ext-workflow/dapr/ext/workflow/_durabletask/internal/orchestrator_actions_pb2.pyi
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -104,28 +104,36 @@ class CreateTimerAction(_message.Message):
NAME_FIELD_NUMBER: _builtins.int
CREATETIMER_FIELD_NUMBER: _builtins.int
EXTERNALEVENT_FIELD_NUMBER: _builtins.int
ACTIVITYRETRY_FIELD_NUMBER: _builtins.int
CHILDWORKFLOWRETRY_FIELD_NUMBER: _builtins.int
name: _builtins.str
@_builtins.property
def fireAt(self) -> _timestamp_pb2.Timestamp: ...
@_builtins.property
def createTimer(self) -> _history_events_pb2.TimerOriginCreateTimer: ...
@_builtins.property
def externalEvent(self) -> _history_events_pb2.TimerOriginExternalEvent: ...
@_builtins.property
def activityRetry(self) -> _history_events_pb2.TimerOriginActivityRetry: ...
@_builtins.property
def childWorkflowRetry(self) -> _history_events_pb2.TimerOriginChildWorkflowRetry: ...
def __init__(
self,
*,
fireAt: _timestamp_pb2.Timestamp | None = ...,
name: _builtins.str | None = ...,
createTimer: _history_events_pb2.TimerOriginCreateTimer | None = ...,
externalEvent: _history_events_pb2.TimerOriginExternalEvent | None = ...,
activityRetry: _history_events_pb2.TimerOriginActivityRetry | None = ...,
childWorkflowRetry: _history_events_pb2.TimerOriginChildWorkflowRetry | None = ...,
) -> None: ...
_HasFieldArgType: _TypeAlias = _typing.Literal["_name", b"_name", "createTimer", b"createTimer", "externalEvent", b"externalEvent", "fireAt", b"fireAt", "name", b"name", "origin", b"origin"] # noqa: Y015
_HasFieldArgType: _TypeAlias = _typing.Literal["_name", b"_name", "activityRetry", b"activityRetry", "childWorkflowRetry", b"childWorkflowRetry", "createTimer", b"createTimer", "externalEvent", b"externalEvent", "fireAt", b"fireAt", "name", b"name", "origin", b"origin"] # noqa: Y015
def HasField(self, field_name: _HasFieldArgType) -> _builtins.bool: ...
_ClearFieldArgType: _TypeAlias = _typing.Literal["_name", b"_name", "createTimer", b"createTimer", "externalEvent", b"externalEvent", "fireAt", b"fireAt", "name", b"name", "origin", b"origin"] # noqa: Y015
_ClearFieldArgType: _TypeAlias = _typing.Literal["_name", b"_name", "activityRetry", b"activityRetry", "childWorkflowRetry", b"childWorkflowRetry", "createTimer", b"createTimer", "externalEvent", b"externalEvent", "fireAt", b"fireAt", "name", b"name", "origin", b"origin"] # noqa: Y015
def ClearField(self, field_name: _ClearFieldArgType) -> None: ...
_WhichOneofReturnType__name: _TypeAlias = _typing.Literal["name"] # noqa: Y015
_WhichOneofArgType__name: _TypeAlias = _typing.Literal["_name", b"_name"] # noqa: Y015
_WhichOneofReturnType_origin: _TypeAlias = _typing.Literal["createTimer", "externalEvent"] # noqa: Y015
_WhichOneofReturnType_origin: _TypeAlias = _typing.Literal["createTimer", "externalEvent", "activityRetry", "childWorkflowRetry"] # noqa: Y015
_WhichOneofArgType_origin: _TypeAlias = _typing.Literal["origin", b"origin"] # noqa: Y015
@_typing.overload
def WhichOneof(self, oneof_group: _WhichOneofArgType__name) -> _WhichOneofReturnType__name | None: ...
Expand Down
Loading
Loading