Skip to content

Commit d72dcb3

Browse files
feiyang3catfeiyang3cat
committed
fix bug in mergeAndUpdate
1 parent 4d67ea5 commit d72dcb3

2 files changed

Lines changed: 169 additions & 37 deletions

File tree

‎service/history/api/updateworkflowoptions/api.go‎

Lines changed: 45 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -172,8 +172,10 @@ func getOptionsFromMutableState(ms historyi.MutableState) *workflowpb.WorkflowEx
172172
opts.Priority = cloned
173173
}
174174
}
175-
if timeSkippingInfo := ms.GetExecutionInfo().GetTimeSkippingInfo(); timeSkippingInfo != nil {
176-
opts.TimeSkippingConfig = timeSkippingInfo.GetConfig()
175+
if tsInfo := ms.GetExecutionInfo().GetTimeSkippingInfo(); tsInfo != nil {
176+
if cloned, ok := proto.Clone(tsInfo.GetConfig()).(*workflowpb.TimeSkippingConfig); ok {
177+
opts.TimeSkippingConfig = cloned
178+
}
177179
}
178180
return opts
179181
}
@@ -241,5 +243,46 @@ func mergeWorkflowExecutionOptions(
241243
}
242244
}
243245

246+
if _, ok := updateFields["timeSkippingConfig.enabled"]; ok {
247+
if mergeInto.TimeSkippingConfig == nil {
248+
mergeInto.TimeSkippingConfig = &workflowpb.TimeSkippingConfig{}
249+
}
250+
mergeInto.TimeSkippingConfig.Enabled = mergeFrom.GetTimeSkippingConfig().GetEnabled()
251+
}
252+
253+
if _, ok := updateFields["timeSkippingConfig.disablePropagation"]; ok {
254+
if mergeInto.TimeSkippingConfig == nil {
255+
mergeInto.TimeSkippingConfig = &workflowpb.TimeSkippingConfig{}
256+
}
257+
mergeInto.TimeSkippingConfig.DisablePropagation = mergeFrom.GetTimeSkippingConfig().GetDisablePropagation()
258+
}
259+
260+
if _, ok := updateFields["timeSkippingConfig.maxSkippedDuration"]; ok {
261+
if mergeInto.TimeSkippingConfig == nil {
262+
mergeInto.TimeSkippingConfig = &workflowpb.TimeSkippingConfig{}
263+
}
264+
mergeInto.TimeSkippingConfig.Bound = &workflowpb.TimeSkippingConfig_MaxSkippedDuration{
265+
MaxSkippedDuration: mergeFrom.GetTimeSkippingConfig().GetMaxSkippedDuration(),
266+
}
267+
}
268+
269+
if _, ok := updateFields["timeSkippingConfig.maxElapsedDuration"]; ok {
270+
if mergeInto.TimeSkippingConfig == nil {
271+
mergeInto.TimeSkippingConfig = &workflowpb.TimeSkippingConfig{}
272+
}
273+
mergeInto.TimeSkippingConfig.Bound = &workflowpb.TimeSkippingConfig_MaxElapsedDuration{
274+
MaxElapsedDuration: mergeFrom.GetTimeSkippingConfig().GetMaxElapsedDuration(),
275+
}
276+
}
277+
278+
if _, ok := updateFields["timeSkippingConfig.maxTargetTime"]; ok {
279+
if mergeInto.TimeSkippingConfig == nil {
280+
mergeInto.TimeSkippingConfig = &workflowpb.TimeSkippingConfig{}
281+
}
282+
mergeInto.TimeSkippingConfig.Bound = &workflowpb.TimeSkippingConfig_MaxTargetTime{
283+
MaxTargetTime: mergeFrom.GetTimeSkippingConfig().GetMaxTargetTime(),
284+
}
285+
}
286+
244287
return mergeInto, nil
245288
}

‎service/history/api/updateworkflowoptions/api_test.go‎

Lines changed: 124 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@ import (
3030
"google.golang.org/protobuf/proto"
3131
"google.golang.org/protobuf/types/known/durationpb"
3232
"google.golang.org/protobuf/types/known/fieldmaskpb"
33+
"google.golang.org/protobuf/types/known/timestamppb"
3334
)
3435

3536
type noopVersionMembershipCache struct{}
@@ -215,41 +216,6 @@ func TestMergeOptions_TimeSkippingConfig(t *testing.T) {
215216
}
216217
}
217218

218-
func TestMergeAndApply_TimeSkippingConfig(t *testing.T) {
219-
tscMask := &fieldmaskpb.FieldMask{Paths: []string{"time_skipping_config"}}
220-
cfg := &workflowpb.TimeSkippingConfig{
221-
Enabled: true,
222-
Bound: &workflowpb.TimeSkippingConfig_MaxSkippedDuration{MaxSkippedDuration: durationpb.New(time.Hour)},
223-
}
224-
225-
t.Run("same config - no event written", func(t *testing.T) {
226-
ctrl := gomock.NewController(t)
227-
ms := historyi.NewMockMutableState(ctrl)
228-
ms.EXPECT().GetExecutionInfo().Return(&persistencespb.WorkflowExecutionInfo{
229-
TimeSkippingInfo: &persistencespb.TimeSkippingInfo{Config: cfg},
230-
}).AnyTimes()
231-
232-
merged, hasChanges, err := MergeAndApply(ms, &workflowpb.WorkflowExecutionOptions{TimeSkippingConfig: cfg}, tscMask, "")
233-
require.NoError(t, err)
234-
require.False(t, hasChanges)
235-
require.True(t, proto.Equal(cfg, merged.GetTimeSkippingConfig()))
236-
})
237-
238-
t.Run("new config - event written with config", func(t *testing.T) {
239-
ctrl := gomock.NewController(t)
240-
ms := historyi.NewMockMutableState(ctrl)
241-
ms.EXPECT().GetExecutionInfo().Return(&persistencespb.WorkflowExecutionInfo{}).AnyTimes()
242-
ms.EXPECT().AddWorkflowExecutionOptionsUpdatedEvent(
243-
nil, true, "", nil, nil, "", nil, cfg,
244-
).Return(&historypb.HistoryEvent{}, nil)
245-
246-
merged, hasChanges, err := MergeAndApply(ms, &workflowpb.WorkflowExecutionOptions{TimeSkippingConfig: cfg}, tscMask, "")
247-
require.NoError(t, err)
248-
require.True(t, hasChanges)
249-
require.True(t, proto.Equal(cfg, merged.GetTimeSkippingConfig()))
250-
})
251-
}
252-
253219
type (
254220
// updateWorkflowOptionsSuite contains tests for the UpdateWorkflowOptions API.
255221
updateWorkflowOptionsSuite struct {
@@ -372,3 +338,126 @@ func (s *updateWorkflowOptionsSuite) TestInvoke_Success() {
372338
s.NotNil(resp)
373339
proto.Equal(expectedOverrideOptions, resp.GetWorkflowExecutionOptions())
374340
}
341+
342+
func TestMergeAndApply_TimeSkippingConfig(t *testing.T) {
343+
oneHour := durationpb.New(time.Hour)
344+
twoHours := durationpb.New(2 * time.Hour)
345+
thirtyMin := durationpb.New(30 * time.Minute)
346+
targetTime := timestamppb.New(time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC))
347+
348+
testCases := []struct {
349+
name string
350+
initialConfig *workflowpb.TimeSkippingConfig
351+
updateOptions *workflowpb.WorkflowExecutionOptions
352+
updateMask *fieldmaskpb.FieldMask
353+
expectedConfig *workflowpb.TimeSkippingConfig
354+
}{
355+
{
356+
name: "update max_skipped_duration preserves enabled and disable_propagation",
357+
initialConfig: &workflowpb.TimeSkippingConfig{
358+
Enabled: true,
359+
DisablePropagation: true,
360+
Bound: &workflowpb.TimeSkippingConfig_MaxSkippedDuration{
361+
MaxSkippedDuration: oneHour,
362+
},
363+
},
364+
updateOptions: &workflowpb.WorkflowExecutionOptions{
365+
TimeSkippingConfig: &workflowpb.TimeSkippingConfig{
366+
Bound: &workflowpb.TimeSkippingConfig_MaxSkippedDuration{
367+
MaxSkippedDuration: twoHours,
368+
},
369+
},
370+
},
371+
updateMask: &fieldmaskpb.FieldMask{Paths: []string{"time_skipping_config.max_skipped_duration"}},
372+
expectedConfig: &workflowpb.TimeSkippingConfig{
373+
Enabled: true,
374+
DisablePropagation: true,
375+
Bound: &workflowpb.TimeSkippingConfig_MaxSkippedDuration{
376+
MaxSkippedDuration: twoHours,
377+
},
378+
},
379+
},
380+
{
381+
name: "change bound type to max_elapsed_duration preserves enabled",
382+
initialConfig: &workflowpb.TimeSkippingConfig{
383+
Enabled: true,
384+
Bound: &workflowpb.TimeSkippingConfig_MaxSkippedDuration{
385+
MaxSkippedDuration: oneHour,
386+
},
387+
},
388+
updateOptions: &workflowpb.WorkflowExecutionOptions{
389+
TimeSkippingConfig: &workflowpb.TimeSkippingConfig{
390+
Bound: &workflowpb.TimeSkippingConfig_MaxElapsedDuration{
391+
MaxElapsedDuration: thirtyMin,
392+
},
393+
},
394+
},
395+
updateMask: &fieldmaskpb.FieldMask{Paths: []string{"time_skipping_config.max_elapsed_duration"}},
396+
expectedConfig: &workflowpb.TimeSkippingConfig{
397+
Enabled: true,
398+
Bound: &workflowpb.TimeSkippingConfig_MaxElapsedDuration{
399+
MaxElapsedDuration: thirtyMin,
400+
},
401+
},
402+
},
403+
{
404+
name: "enable from nil config",
405+
initialConfig: nil,
406+
updateOptions: &workflowpb.WorkflowExecutionOptions{
407+
TimeSkippingConfig: &workflowpb.TimeSkippingConfig{Enabled: true},
408+
},
409+
updateMask: &fieldmaskpb.FieldMask{Paths: []string{"time_skipping_config.enabled"}},
410+
expectedConfig: &workflowpb.TimeSkippingConfig{Enabled: true},
411+
},
412+
{
413+
name: "disable from enabled config",
414+
initialConfig: &workflowpb.TimeSkippingConfig{Enabled: true},
415+
updateOptions: &workflowpb.WorkflowExecutionOptions{
416+
TimeSkippingConfig: &workflowpb.TimeSkippingConfig{Enabled: false},
417+
},
418+
updateMask: &fieldmaskpb.FieldMask{Paths: []string{"time_skipping_config.enabled"}},
419+
expectedConfig: &workflowpb.TimeSkippingConfig{Enabled: false},
420+
},
421+
{
422+
name: "change bound type to max_target_time preserves enabled",
423+
initialConfig: &workflowpb.TimeSkippingConfig{
424+
Enabled: true,
425+
Bound: &workflowpb.TimeSkippingConfig_MaxSkippedDuration{
426+
MaxSkippedDuration: oneHour,
427+
},
428+
},
429+
updateOptions: &workflowpb.WorkflowExecutionOptions{
430+
TimeSkippingConfig: &workflowpb.TimeSkippingConfig{
431+
Bound: &workflowpb.TimeSkippingConfig_MaxTargetTime{
432+
MaxTargetTime: targetTime,
433+
},
434+
},
435+
},
436+
updateMask: &fieldmaskpb.FieldMask{Paths: []string{"time_skipping_config.max_target_time"}},
437+
expectedConfig: &workflowpb.TimeSkippingConfig{
438+
Enabled: true,
439+
Bound: &workflowpb.TimeSkippingConfig_MaxTargetTime{
440+
MaxTargetTime: targetTime,
441+
},
442+
},
443+
},
444+
}
445+
446+
for _, tc := range testCases {
447+
t.Run(tc.name, func(t *testing.T) {
448+
ctrl := gomock.NewController(t)
449+
ms := historyi.NewMockMutableState(ctrl)
450+
ms.EXPECT().GetExecutionInfo().Return(&persistencespb.WorkflowExecutionInfo{
451+
TimeSkippingInfo: &persistencespb.TimeSkippingInfo{
452+
Config: tc.initialConfig,
453+
},
454+
}).AnyTimes()
455+
ms.EXPECT().AddWorkflowExecutionOptionsUpdatedEvent(nil, true, "", nil, nil, "", nil, gomock.Any()).Return(&historypb.HistoryEvent{}, nil)
456+
457+
result, hasChanges, err := MergeAndApply(ms, tc.updateOptions, tc.updateMask, "")
458+
require.NoError(t, err)
459+
require.True(t, hasChanges)
460+
require.True(t, proto.Equal(tc.expectedConfig, result.GetTimeSkippingConfig()))
461+
})
462+
}
463+
}

0 commit comments

Comments
 (0)