Skip to content

Commit 1000c5e

Browse files
feiyang3catfeiyang3cat
committed
fix the bug in fast forwarding cap
1 parent b913aec commit 1000c5e

20 files changed

Lines changed: 258 additions & 241 deletions

‎api/persistence/v1/executions.pb.go‎

Lines changed: 10 additions & 10 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎cmd/tools/getproto/files.go‎

Lines changed: 0 additions & 1 deletion
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎go.mod‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -64,7 +64,7 @@ require (
6464
go.opentelemetry.io/otel/sdk v1.43.0
6565
go.opentelemetry.io/otel/sdk/metric v1.43.0
6666
go.opentelemetry.io/otel/trace v1.43.0
67-
go.temporal.io/api v1.62.14-0.20260609212628-1584598228d8
67+
go.temporal.io/api v1.62.14-0.20260610035406-3eef34cf314c
6868
go.temporal.io/auto-scaled-workers v0.0.0-20260407181057-edd947d743d2
6969
go.temporal.io/sdk v1.41.1
7070
go.uber.org/fx v1.24.0

‎go.sum‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -471,8 +471,8 @@ go.opentelemetry.io/proto/slim/otlp/collector/profiles/v1development v0.3.0 h1:R
471471
go.opentelemetry.io/proto/slim/otlp/collector/profiles/v1development v0.3.0/go.mod h1:I89cynRj8y+383o7tEQVg2SVA6SRgDVIouWPUVXjx0U=
472472
go.opentelemetry.io/proto/slim/otlp/profiles/v1development v0.3.0 h1:CQvJSldHRUN6Z8jsUeYv8J0lXRvygALXIzsmAeCcZE0=
473473
go.opentelemetry.io/proto/slim/otlp/profiles/v1development v0.3.0/go.mod h1:xSQ+mEfJe/GjK1LXEyVOoSI1N9JV9ZI923X5kup43W4=
474-
go.temporal.io/api v1.62.14-0.20260609212628-1584598228d8 h1:i3eIlnuEd9KoKkQVKDcx3+Vv/fYNpmNo0cIc/MlWzzs=
475-
go.temporal.io/api v1.62.14-0.20260609212628-1584598228d8/go.mod h1:0k75tRljEuELWGeXjEZZO7zYqBln4+1FrG6+IMOMy7Q=
474+
go.temporal.io/api v1.62.14-0.20260610035406-3eef34cf314c h1:F77ikVukHHPvNBQ48mt6OUGwXEIzSznXc4FHlia6l9I=
475+
go.temporal.io/api v1.62.14-0.20260610035406-3eef34cf314c/go.mod h1:0k75tRljEuELWGeXjEZZO7zYqBln4+1FrG6+IMOMy7Q=
476476
go.temporal.io/auto-scaled-workers v0.0.0-20260407181057-edd947d743d2 h1:1hKeH3GyR6YD6LKMHGCZ76t6h1Sgha0hXVQBxWi3dlQ=
477477
go.temporal.io/auto-scaled-workers v0.0.0-20260407181057-edd947d743d2/go.mod h1:T8dnzVPeO+gaUTj9eDgm/lT2lZH4+JXNvrGaQGyVi50=
478478
go.temporal.io/sdk v1.41.1 h1:yOpvsHyDD1lNuwlGBv/SUodCPhjv9nDeC9lLHW/fJUA=

‎proto/internal/temporal/server/api/historyservice/v1/request_response.proto‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -108,7 +108,6 @@ message StartWorkflowExecutionRequest {
108108
temporal.api.workflow.v1.TimeSkippingStatePropagation time_skipping_state_propagation = 19;
109109
}
110110

111-
112111
message StartWorkflowExecutionResponse {
113112
string run_id = 1;
114113
temporal.server.api.clock.v1.VectorClock clock = 2;

‎proto/internal/temporal/server/api/persistence/v1/executions.proto‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -330,8 +330,8 @@ message TimeSkippingInfo {
330330
// Total skipped duration for the current workflow execution run, including any
331331
./* inherited skipped duration carried over from a preceding execution that started this run. */google.protobuf.Duration accumulated_skipped_duration = 2;
332332

333-
// The current fast-forward for time skipping.
334-
FastForwardInfo fast_forward = 4;
333+
// The current fast-forward info for time skipping.
334+
FastForwardInfo fast_forward_info = 4;
335335
}
336336

337337
message FastForwardInfo {

‎service/history/historybuilder/event_factory.go‎

Lines changed: 9 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -85,6 +85,7 @@ func (b *EventFactory) CreateWorkflowExecutionStartedEvent(
8585
InheritedAutoUpgradeInfo: request.InheritedAutoUpgradeInfo,
8686
DeclinedTargetVersionUpgrade: request.DeclinedTargetVersionUpgrade,
8787
TimeSkippingConfig: req.GetTimeSkippingConfig(),
88+
TimeSkippingStatePropagation: request.GetTimeSkippingStatePropagation(),
8889
}
8990

9091
parentInfo := request.ParentExecutionInfo
@@ -843,7 +844,7 @@ func (b *EventFactory) CreateStartChildWorkflowExecutionInitiatedEvent(
843844
command *commandpb.StartChildWorkflowExecutionCommandAttributes,
844845
targetNamespaceID namespace.ID,
845846
timeSkippingConfig *workflowpb.TimeSkippingConfig,
846-
initialSkippedDuration *durationpb.Duration,
847+
statePropagation *workflowpb.TimeSkippingStatePropagation,
847848
) *historypb.HistoryEvent {
848849
event := b.createHistoryEvent(enumspb.EVENT_TYPE_START_CHILD_WORKFLOW_EXECUTION_INITIATED, b.timeSource.Now())
849850
event.Attributes = &historypb.HistoryEvent_StartChildWorkflowExecutionInitiatedEventAttributes{
@@ -866,13 +867,13 @@ func (b *EventFactory) CreateStartChildWorkflowExecutionInitiatedEvent(
866867
// Filter nil values here rather than in the API layer because not all
867868
// creation paths go through the frontend (continue-as-new, child workflows, replication).
868869
// This CaN event is created on the parent workflow, so we need to filter nil values here.
869-
Memo: payload.FilterNilMemo(command.Memo),
870-
SearchAttributes: payload.FilterNilSearchAttributes(command.SearchAttributes),
871-
ParentClosePolicy: command.GetParentClosePolicy(),
872-
InheritBuildId: command.InheritBuildId, //nolint:staticcheck // SA1019: worker versioning v0.2
873-
Priority: command.Priority,
874-
TimeSkippingConfig: timeSkippingConfig,
875-
InitialSkippedDuration: initialSkippedDuration,
870+
Memo: payload.FilterNilMemo(command.Memo),
871+
SearchAttributes: payload.FilterNilSearchAttributes(command.SearchAttributes),
872+
ParentClosePolicy: command.GetParentClosePolicy(),
873+
InheritBuildId: command.InheritBuildId, //nolint:staticcheck // SA1019: worker versioning v0.2
874+
Priority: command.Priority,
875+
TimeSkippingConfig: timeSkippingConfig,
876+
TimeSkippingStatePropagation: statePropagation,
876877
},
877878
}
878879
return event

‎service/history/historybuilder/history_builder.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -770,14 +770,14 @@ func (b *HistoryBuilder) AddStartChildWorkflowExecutionInitiatedEvent(
770770
command *commandpb.StartChildWorkflowExecutionCommandAttributes,
771771
targetNamespaceID namespace.ID,
772772
timeSkippingConfig *workflowpb.TimeSkippingConfig,
773-
initialSkippedDuration *durationpb.Duration,
773+
statePropagation *workflowpb.TimeSkippingStatePropagation,
774774
) (*historypb.HistoryEvent, int64) {
775775
event := b.EventFactory.CreateStartChildWorkflowExecutionInitiatedEvent(
776776
workflowTaskCompletedEventID,
777777
command,
778778
targetNamespaceID,
779779
timeSkippingConfig,
780-
initialSkippedDuration,
780+
statePropagation,
781781
)
782782
return b.add(event)
783783
}

‎service/history/timer_queue_active_task_executor.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -954,7 +954,7 @@ func fastForwardTaskIsLive(mutableState historyi.MutableState, task *tasks.TimeS
954954
if tsi == nil || !tsi.GetConfig().GetEnabled() {
955955
return false
956956
}
957-
fastForward := tsi.GetFastForward()
957+
fastForward := tsi.GetFastForwardInfo()
958958
if fastForward == nil || fastForward.GetTargetTime() == nil || fastForward.GetSourceEventId() == 0 || fastForward.GetHasReached() {
959959
return false
960960
}

‎service/history/timer_queue_active_task_executor_test.go‎

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -2417,7 +2417,7 @@ func (s *timerQueueActiveTaskExecutorSuite) TestExecuteTimeSkippingTimerTask_Wor
24172417
Config: &workflowpb.TimeSkippingConfig{
24182418
Enabled: true,
24192419
FastForward: durationpb.New(time.Hour)},
2420-
FastForward: &persistencespb.FastForwardInfo{
2420+
FastForwardInfo: &persistencespb.FastForwardInfo{
24212421
TargetTime: timestamppb.New(s.now.Add(time.Hour)),
24222422
SourceEventId: 1,
24232423
},
@@ -2457,7 +2457,7 @@ func (s *timerQueueActiveTaskExecutorSuite) TestExecuteTimeSkippingTimerTask_Con
24572457
pms, workflowKey := s.makeTimeSkippingMS()
24582458
pms.ExecutionInfo.TimeSkippingInfo = &persistencespb.TimeSkippingInfo{
24592459
Config: &workflowpb.TimeSkippingConfig{Enabled: false},
2460-
FastForward: &persistencespb.FastForwardInfo{
2460+
FastForwardInfo: &persistencespb.FastForwardInfo{
24612461
TargetTime: timestamppb.New(s.now.Add(time.Hour)),
24622462
SourceEventId: 1,
24632463
},
@@ -2500,7 +2500,7 @@ func (s *timerQueueActiveTaskExecutorSuite) TestExecuteTimeSkippingTimerTask_Has
25002500
pms, workflowKey := s.makeTimeSkippingMS()
25012501
pms.ExecutionInfo.TimeSkippingInfo = &persistencespb.TimeSkippingInfo{
25022502
Config: &workflowpb.TimeSkippingConfig{Enabled: true},
2503-
FastForward: &persistencespb.FastForwardInfo{
2503+
FastForwardInfo: &persistencespb.FastForwardInfo{
25042504
TargetTime: timestamppb.New(s.now.Add(time.Hour)),
25052505
SourceEventId: 1,
25062506
HasReached: true,
@@ -2524,7 +2524,7 @@ func (s *timerQueueActiveTaskExecutorSuite) TestExecuteTimeSkippingTimerTask_Sou
25242524
pms, workflowKey := s.makeTimeSkippingMS()
25252525
pms.ExecutionInfo.TimeSkippingInfo = &persistencespb.TimeSkippingInfo{
25262526
Config: &workflowpb.TimeSkippingConfig{Enabled: true},
2527-
FastForward: &persistencespb.FastForwardInfo{
2527+
FastForwardInfo: &persistencespb.FastForwardInfo{
25282528
TargetTime: timestamppb.New(s.now.Add(time.Hour)),
25292529
SourceEventId: 0,
25302530
},
@@ -2547,7 +2547,7 @@ func (s *timerQueueActiveTaskExecutorSuite) TestExecuteTimeSkippingTimerTask_Sou
25472547
pms, workflowKey := s.makeTimeSkippingMS()
25482548
pms.ExecutionInfo.TimeSkippingInfo = &persistencespb.TimeSkippingInfo{
25492549
Config: &workflowpb.TimeSkippingConfig{Enabled: true},
2550-
FastForward: &persistencespb.FastForwardInfo{
2550+
FastForwardInfo: &persistencespb.FastForwardInfo{
25512551
TargetTime: timestamppb.New(s.now.Add(time.Hour)),
25522552
SourceEventId: 2, // fast-forward was reconfigured; old task carries EventID=1.
25532553
},
@@ -2572,7 +2572,7 @@ func (s *timerQueueActiveTaskExecutorSuite) TestExecuteTimeSkippingTimerTask_Hap
25722572
Config: &workflowpb.TimeSkippingConfig{
25732573
Enabled: true,
25742574
FastForward: durationpb.New(time.Hour)},
2575-
FastForward: &persistencespb.FastForwardInfo{
2575+
FastForwardInfo: &persistencespb.FastForwardInfo{
25762576
TargetTime: timestamppb.New(s.now.Add(time.Hour)),
25772577
SourceEventId: 1,
25782578
},
@@ -2598,7 +2598,7 @@ func (s *timerQueueActiveTaskExecutorSuite) TestExecuteTimeSkippingTimerTask_Hap
25982598
tsi := loaded.GetExecutionInfo().GetTimeSkippingInfo()
25992599
s.Require().NotNil(tsi)
26002600
s.False(tsi.GetConfig().GetEnabled(), "happy path must flip Enabled=false")
2601-
s.True(tsi.GetFastForward().GetHasReached(), "happy path must set HasReached=true")
2601+
s.True(tsi.GetFastForwardInfo().GetHasReached(), "happy path must set HasReached=true")
26022602
}
26032603

26042604
func (s *timerQueueActiveTaskExecutorSuite) TestProcessSingleActivityTimeoutTask() {

0 commit comments

Comments
 (0)