Skip to content

(WIP) TS: fast forward and propagation policy change #10576

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
feiyang3cat wants to merge 11 commits into
base: main
Choose a base branch
from
Open

(WIP) TS: fast forward and propagation policy change #10576

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
941012c
make compatible to time skipping api changes
feiyang3cat Jun 6, 2026
49fccaa
update propagation policy and renaming time skipping bound to fast
feiyang3cat Jun 8, 2026
71998a6
add time skipping utils
feiyang3cat Jun 9, 2026
ba3f3fb
add testing cases for propagation
feiyang3cat Jun 10, 2026
af0438e
add TimeSkippingPropagation to api
feiyang3cat Jun 10, 2026
7282f2b
fix the bug in fast forwarding cap
feiyang3cat Jun 10, 2026
e1a161e
fix propagation of virtual time to child wf
feiyang3cat Jun 11, 2026
81f4c68
update api
feiyang3cat Jun 11, 2026
64eee9a
refine tests
feiyang3cat Jun 12, 2026
8e62251
refine propagation tests
feiyang3cat Jun 12, 2026
e54d182
accomodate the api change
feiyang3cat Jun 13, 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
657 changes: 334 additions & 323 deletions api/historyservice/v1/request_response.pb.go
View file
Open in desktop

Large diffs are not rendered by default.

20 changes: 10 additions & 10 deletions api/persistence/v1/executions.go-helpers.pb.go
View file
Open in desktop

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

60 changes: 30 additions & 30 deletions api/persistence/v1/executions.pb.go
View file
Open in desktop

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

2 changes: 1 addition & 1 deletion go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,7 +64,7 @@ require (
go.opentelemetry.io/otel/sdk v1.43.0
go.opentelemetry.io/otel/sdk/metric v1.43.0
go.opentelemetry.io/otel/trace v1.43.0
go.temporal.io/api v1.62.14-0.20260610204950-15ceffd37b1b
go.temporal.io/api v1.62.14-0.20260612233023-cc33a0049faf
go.temporal.io/auto-scaled-workers v0.0.0-20260407181057-edd947d743d2
go.temporal.io/sdk v1.41.1
go.uber.org/fx v1.24.0
Expand Down
6 changes: 4 additions & 2 deletions go.sum
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -471,8 +471,10 @@ go.opentelemetry.io/proto/slim/otlp/collector/profiles/v1development v0.3.0 h1:R
go.opentelemetry.io/proto/slim/otlp/collector/profiles/v1development v0.3.0/go.mod h1:I89cynRj8y+383o7tEQVg2SVA6SRgDVIouWPUVXjx0U=
go.opentelemetry.io/proto/slim/otlp/profiles/v1development v0.3.0 h1:CQvJSldHRUN6Z8jsUeYv8J0lXRvygALXIzsmAeCcZE0=
go.opentelemetry.io/proto/slim/otlp/profiles/v1development v0.3.0/go.mod h1:xSQ+mEfJe/GjK1LXEyVOoSI1N9JV9ZI923X5kup43W4=
go.temporal.io/api v1.62.14-0.20260610204950-15ceffd37b1b h1:cm1d0DOh5VXiEZC0QEjxuy7ZtK//4H+K+bYpV7B2gV4=
go.temporal.io/api v1.62.14-0.20260610204950-15ceffd37b1b/go.mod h1:0k75tRljEuELWGeXjEZZO7zYqBln4+1FrG6+IMOMy7Q=
go.temporal.io/api v1.62.14-0.20260611214622-59fad4392578 h1:Gu3lclbgjOwc4/KPPckFqXwb+kwz6RTtvwaf21SmU+g=
go.temporal.io/api v1.62.14-0.20260611214622-59fad4392578/go.mod h1:0k75tRljEuELWGeXjEZZO7zYqBln4+1FrG6+IMOMy7Q=
go.temporal.io/api v1.62.14-0.20260612233023-cc33a0049faf h1:aZn5SIyAKmtg/sfu2FOajT65RRu3ksGQ9/8vphd4O4A=
go.temporal.io/api v1.62.14-0.20260612233023-cc33a0049faf/go.mod h1:0k75tRljEuELWGeXjEZZO7zYqBln4+1FrG6+IMOMy7Q=
go.temporal.io/auto-scaled-workers v0.0.0-20260407181057-edd947d743d2 h1:1hKeH3GyR6YD6LKMHGCZ76t6h1Sgha0hXVQBxWi3dlQ=
go.temporal.io/auto-scaled-workers v0.0.0-20260407181057-edd947d743d2/go.mod h1:T8dnzVPeO+gaUTj9eDgm/lT2lZH4+JXNvrGaQGyVi50=
go.temporal.io/sdk v1.41.1 h1:yOpvsHyDD1lNuwlGBv/SUodCPhjv9nDeC9lLHW/fJUA=
Expand Down
10 changes: 5 additions & 5 deletions proto/internal/temporal/server/api/historyservice/v1/request_response.proto
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -104,11 +104,8 @@ message StartWorkflowExecutionRequest {
// directly from the started event. Written onto the new run's
// WorkflowExecutionStartedEvent.
temporal.api.history.v1.DeclinedTargetVersionUpgrade declined_target_version_upgrade = 17;

// If a workflow execution is started by a previous execution (parent-child workflow or continue-as-new)
// that has already skipped some time, the accumulated skipped duration from that execution
// can be passed to the new workflow execution through this field.
google.protobuf.Duration initial_skipped_duration = 18;
reserved 18;
temporal.api.common.v1.TimeSkippingStatePropagation time_skipping_state_propagation = 19;
}

message StartWorkflowExecutionResponse {
Expand Down Expand Up @@ -1384,6 +1381,9 @@ message UpdateWorkflowExecutionOptionsRequest {
message UpdateWorkflowExecutionOptionsResponse {
// Workflow Execution options after update.
temporal.api.workflow.v1.WorkflowExecutionOptions workflow_execution_options = 1;
// Time of the workflow execution at the moment the options were applied.
// If time skipping happened in this execution, it will return virtual time.
google.protobuf.Timestamp update_time = 2;
}

message PauseWorkflowExecutionRequest {
Expand Down
17 changes: 10 additions & 7 deletions proto/internal/temporal/server/api/persistence/v1/executions.proto
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -321,27 +321,30 @@ message WorkflowExecutionInfo {
}

message TimeSkippingInfo {
reserved 3;
reserved "current_elapsed_duration_bound";

// Current time-skipping configuration applied to the workflow.
temporal.api.workflow.v1.TimeSkippingConfig config = 1;
temporal.api.common.v1.TimeSkippingConfig config = 1;

// Total skipped duration for the current workflow execution run, including any
./* inherited skipped duration carried over from a preceding execution that started this run. */google.protobuf.Duration accumulated_skipped_duration = 2;

// The current bound based on elapsed duration for time skipping.
TimeSkippingBoundInfo current_elapsed_duration_bound = 3;
// The current fast-forward info for time skipping.
FastForwardInfo fast_forward_info = 4;
}

message TimeSkippingBoundInfo {
// Target time for a bound, expressed in virtual time.
message FastForwardInfo {
// Target time for the fast-forward, expressed in virtual time.
google.protobuf.Timestamp target_time = 1;

// Indicates whether this bound has already been reached, used for idempotency checks.
// Indicates whether this fast-forward has already been reached, used for idempotency checks.
bool has_reached = 2;

// Event ID of the WorkflowExecutionStartedEvent (always 1) or the most recent
// WorkflowExecutionOptionsUpdatedEvent that introduced the current time-skipping
// configuration. This is used as the task event ID for the time-skipping timer task,
// enabling reset and replication-conflict resolution to identify obsolete bound tasks
// enabling reset and replication-conflict resolution to identify obsolete fast-forward tasks
// via the standard staleness check.
int64 source_event_id = 3;
}
Expand Down
12 changes: 5 additions & 7 deletions service/frontend/workflow_handler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -700,7 +700,7 @@ func (wh *WorkflowHandler) prepareStartWorkflowRequest(
}

func (wh *WorkflowHandler) validateTimeSkippingConfig(
tsc *workflowpb.TimeSkippingConfig,
tsc *commonpb.TimeSkippingConfig,
ns namespace.Name,
) error {
if tsc == nil {
Expand All @@ -715,15 +715,13 @@ func (wh *WorkflowHandler) validateTimeSkippingConfig(
}

if !tsc.GetEnabled() {
if tsc.GetBound() != nil {
return serviceerror.NewInvalidArgument("time_skipping_config: cannot set bound when enabled is false")
if tsc.GetFastForward() != nil {
return serviceerror.NewInvalidArgument("time_skipping_config: cannot set fast_forward when enabled is false")
}
return nil
}
if b, ok := tsc.GetBound().(*workflowpb.TimeSkippingConfig_MaxElapsedDuration); ok {
if b.MaxElapsedDuration.AsDuration() < 0 {
return serviceerror.NewInvalidArgument("time_skipping_config: max_elapsed_duration must be positive")
}
if ff := tsc.GetFastForward(); ff != nil && ff.AsDuration() < 0 {
return serviceerror.NewInvalidArgument("time_skipping_config: fast_forward must be positive")
}

return nil
Expand Down
Loading
Loading