go.temporal.io/server/tests/timeskipping_test.go
1315 LOC · 0 covered · 1315 uncovered · 0 ranges · 0 concepts · 0 introducers · 0 tests
1
package tests
2
3
import (
4
"errors"
5
"testing"
6
"time"
7
8
"github.com/google/uuid"
9
commandpb "go.temporal.io/api/command/v1"
10
commonpb "go.temporal.io/api/common/v1"
11
enumspb "go.temporal.io/api/enums/v1"
12
historypb "go.temporal.io/api/history/v1"
13
taskqueuepb "go.temporal.io/api/taskqueue/v1"
14
updatepb "go.temporal.io/api/update/v1"
15
workflowpb "go.temporal.io/api/workflow/v1"
16
"go.temporal.io/api/workflowservice/v1"
17
"go.temporal.io/sdk/converter"
18
sdktemporal "go.temporal.io/sdk/temporal"
19
"go.temporal.io/sdk/workflow"
20
enumsspb "go.temporal.io/server/api/enums/v1"
21
"go.temporal.io/server/chasm"
22
"go.temporal.io/server/common"
23
"go.temporal.io/server/common/dynamicconfig"
24
"go.temporal.io/server/common/persistence"
25
"go.temporal.io/server/common/testing/parallelsuite"
26
"go.temporal.io/server/common/testing/taskpoller"
27
"go.temporal.io/server/common/testing/testvars"
28
historytasks "go.temporal.io/server/service/history/tasks"
29
"go.temporal.io/server/tests/testcore"
30
"google.golang.org/protobuf/proto"
31
"google.golang.org/protobuf/types/known/durationpb"
32
"google.golang.org/protobuf/types/known/fieldmaskpb"
33
)
34
35
type TimeSkippingTestSuite struct {
36
parallelsuite.Suite[*TimeSkippingTestSuite]
37
}
38
39
func TestTimeSkippingTestSuite(t *testing.T) {
40
parallelsuite.Run(t, &TimeSkippingTestSuite{})
41
}
42
43
// TestTimeSkipping_FeatureDisabled verifies that starting a workflow with time skipping
44
// returns an error when the feature flag is off for the namespace.
45
func (s *TimeSkippingTestSuite) TestTimeSkipping_FeatureDisabled() {
46
env := testcore.NewEnv(s.T())
47
// TimeSkippingEnabled defaults to false; no override needed.
48
id := "functional-timeskipping-feature-disabled"
49
tl := "functional-timeskipping-feature-disabled-tq"
50
51
_, err := env.FrontendClient().StartWorkflowExecution(s.Context(), &workflowservice.StartWorkflowExecutionRequest{
52
RequestId: uuid.NewString(),
53
Namespace: env.Namespace().String(),
54
WorkflowId: id,
55
WorkflowType: &commonpb.WorkflowType{Name: id + "-type"},
56
TaskQueue: &taskqueuepb.TaskQueue{Name: tl, Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
57
WorkflowRunTimeout: durationpb.New(100 * time.Second),
58
WorkflowTaskTimeout: durationpb.New(10 * time.Second),
59
TimeSkippingConfig: &commonpb.TimeSkippingConfig{Enabled: true},
60
})
61
s.Error(err, "expected error when time skipping is disabled for namespace")
62
}
63
64
// TestTimeSkipping_StartWorkflow_DCEnabled verifies that StartWorkflowExecution with
65
// TimeSkippingConfig persists the config in mutable state when the feature flag is on.
66
func (s *TimeSkippingTestSuite) TestTimeSkipping_StartWorkflow_DCEnabled() {
67
env := testcore.NewEnv(s.T())
68
env.OverrideDynamicConfig(dynamicconfig.TimeSkippingEnabled, true)
69
tv := testvars.New(s.T())
70
71
inputConfig := &commonpb.TimeSkippingConfig{
72
Enabled: true,
73
FastForward: durationpb.New(time.Hour),
74
}
75
76
resp, err := env.FrontendClient().StartWorkflowExecution(s.Context(), &workflowservice.StartWorkflowExecutionRequest{
77
RequestId: uuid.NewString(),
78
Namespace: env.Namespace().String(),
79
WorkflowId: tv.WorkflowID(),
80
WorkflowType: tv.WorkflowType(),
81
TaskQueue: tv.TaskQueue(),
82
WorkflowRunTimeout: durationpb.New(100 * time.Second),
83
WorkflowTaskTimeout: durationpb.New(10 * time.Second),
84
TimeSkippingConfig: inputConfig,
85
})
86
s.NoError(err)
87
88
ms := s.getMutableState(env, tv.WorkflowID(), resp.RunId)
89
s.True(ms.State.ExecutionInfo.GetTimeSkippingInfo().GetConfig().GetEnabled())
90
s.True(proto.Equal(inputConfig, ms.State.ExecutionInfo.GetTimeSkippingInfo().GetConfig()))
91
}
92
93
// TestTimeSkipping_SignalWithStart_DCEnabled verifies that SignalWithStartWorkflowExecution
94
// with TimeSkippingConfig persists the config in mutable state when the feature flag is on.
95
func (s *TimeSkippingTestSuite) TestTimeSkipping_SignalWithStart_DCEnabled() {
96
env := testcore.NewEnv(s.T())
97
env.OverrideDynamicConfig(dynamicconfig.TimeSkippingEnabled, true)
98
tv := testvars.New(s.T())
99
100
inputConfig := &commonpb.TimeSkippingConfig{
101
Enabled: true,
102
FastForward: durationpb.New(time.Hour),
103
}
104
105
resp, err := env.FrontendClient().SignalWithStartWorkflowExecution(s.Context(), &workflowservice.SignalWithStartWorkflowExecutionRequest{
106
RequestId: uuid.NewString(),
107
Namespace: env.Namespace().String(),
108
WorkflowId: tv.WorkflowID(),
109
WorkflowType: tv.WorkflowType(),
110
TaskQueue: tv.TaskQueue(),
111
WorkflowRunTimeout: durationpb.New(100 * time.Second),
112
WorkflowTaskTimeout: durationpb.New(10 * time.Second),
113
SignalName: tv.SignalName(),
114
TimeSkippingConfig: inputConfig,
115
})
116
s.NoError(err)
117
118
ms := s.getMutableState(env, tv.WorkflowID(), resp.RunId)
119
s.True(proto.Equal(inputConfig, ms.State.ExecutionInfo.GetTimeSkippingInfo().GetConfig()))
120
}
121
122
// TestTimeSkipping_ExecuteMultiOperation_DCEnabled verifies that a StartWorkflow inside
123
// ExecuteMultiOperation with TimeSkippingConfig persists the config in mutable state
124
// when the feature flag is on.
125
func (s *TimeSkippingTestSuite) TestTimeSkipping_ExecuteMultiOperation_DCEnabled() {
126
env := testcore.NewEnv(s.T())
127
env.OverrideDynamicConfig(dynamicconfig.TimeSkippingEnabled, true)
128
tv := testvars.New(s.T())
129
maxElapsedDuration := time.Hour
130
131
inputConfig := &commonpb.TimeSkippingConfig{
132
Enabled: true,
133
FastForward: durationpb.New(maxElapsedDuration),
134
}
135
136
resp, err := env.FrontendClient().ExecuteMultiOperation(s.Context(), &workflowservice.ExecuteMultiOperationRequest{
137
Namespace: env.Namespace().String(),
138
Operations: []*workflowservice.ExecuteMultiOperationRequest_Operation{
139
{
140
Operation: &workflowservice.ExecuteMultiOperationRequest_Operation_StartWorkflow{
141
StartWorkflow: &workflowservice.StartWorkflowExecutionRequest{
142
RequestId: uuid.NewString(),
143
Namespace: env.Namespace().String(),
144
WorkflowId: tv.WorkflowID(),
145
WorkflowType: tv.WorkflowType(),
146
TaskQueue: tv.TaskQueue(),
147
WorkflowRunTimeout: durationpb.New(100 * time.Second),
148
WorkflowTaskTimeout: durationpb.New(10 * time.Second),
149
TimeSkippingConfig: inputConfig,
150
},
151
},
152
},
153
{
154
Operation: &workflowservice.ExecuteMultiOperationRequest_Operation_UpdateWorkflow{
155
UpdateWorkflow: &workflowservice.UpdateWorkflowExecutionRequest{
156
Namespace: env.Namespace().String(),
157
WorkflowExecution: &commonpb.WorkflowExecution{WorkflowId: tv.WorkflowID()},
158
Request: &updatepb.Request{
159
Meta: &updatepb.Meta{UpdateId: uuid.NewString()},
160
Input: &updatepb.Input{Name: "my-update"},
161
},
162
},
163
},
164
},
165
},
166
})
167
s.NoError(err)
168
169
runID := resp.GetResponses()[0].GetStartWorkflow().GetRunId()
170
ms := s.getMutableState(env, tv.WorkflowID(), runID)
171
s.True(proto.Equal(inputConfig, ms.State.ExecutionInfo.GetTimeSkippingInfo().GetConfig()))
172
}
173
174
// TestTimeSkipping_UpdateWorkflowOptions_DCEnabled exercises the full UpdateWorkflowExecutionOptions
175
// lifecycle for TimeSkippingConfig:
176
// 1. Start workflow with no time-skipping — assert mutable state has no config.
177
// 2. First update: enable with max_elapsed_duration — check MS and event 1 attributes.
178
// 3. Second update: change the max_elapsed_duration value — check MS and event 2 attributes.
179
// 4. Third update: disable (Enabled=false) — check MS and event 3 attributes.
180
// 5. Assert exactly 3 WorkflowExecutionOptionsUpdated events appear in history.
181
func (s *TimeSkippingTestSuite) TestTimeSkipping_UpdateWorkflowOptions_DCEnabled() {
182
env := testcore.NewEnv(s.T())
183
env.OverrideDynamicConfig(dynamicconfig.TimeSkippingEnabled, true)
184
tv := testvars.New(s.T())
185
186
// Start a workflow without any time-skipping config.
187
startResp, err := env.FrontendClient().StartWorkflowExecution(s.Context(), &workflowservice.StartWorkflowExecutionRequest{
188
RequestId: uuid.NewString(),
189
Namespace: env.Namespace().String(),
190
WorkflowId: tv.WorkflowID(),
191
WorkflowType: tv.WorkflowType(),
192
TaskQueue: tv.TaskQueue(),
193
WorkflowRunTimeout: durationpb.New(100 * time.Second),
194
WorkflowTaskTimeout: durationpb.New(10 * time.Second),
195
})
196
s.NoError(err)
197
runID := startResp.RunId
198
199
// collectOptionsEvents returns all WorkflowExecutionOptionsUpdated events in history order.
200
collectOptionsEvents := func() []*historypb.HistoryEvent {
201
histResp, err := env.FrontendClient().GetWorkflowExecutionHistory(s.Context(), &workflowservice.GetWorkflowExecutionHistoryRequest{
202
Namespace: env.Namespace().String(),
203
Execution: &commonpb.WorkflowExecution{WorkflowId: tv.WorkflowID(), RunId: runID},
204
})
205
s.NoError(err)
206
var events []*historypb.HistoryEvent
207
for _, e := range histResp.History.Events {
208
if e.EventType == enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_OPTIONS_UPDATED {
209
events = append(events, e)
210
}
211
}
212
return events
213
}
214
updateOptions := func(cfg *commonpb.TimeSkippingConfig) {
215
_, err := env.FrontendClient().UpdateWorkflowExecutionOptions(s.Context(), &workflowservice.UpdateWorkflowExecutionOptionsRequest{
216
Namespace: env.Namespace().String(),
217
WorkflowExecution: &commonpb.WorkflowExecution{WorkflowId: tv.WorkflowID(), RunId: runID},
218
WorkflowExecutionOptions: &workflowpb.WorkflowExecutionOptions{TimeSkippingConfig: cfg},
219
UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"time_skipping_config"}},
220
})
221
s.NoError(err)
222
}
223
224
// No time-skipping config before any update.
225
ms := s.getMutableState(env, tv.WorkflowID(), runID)
226
s.Nil(ms.State.ExecutionInfo.GetTimeSkippingInfo().GetConfig())
227
228
// First update: enable with a max_elapsed_duration.
229
config1 := &commonpb.TimeSkippingConfig{
230
Enabled: true,
231
FastForward: durationpb.New(time.Hour),
232
}
233
updateOptions(config1)
234
235
ms = s.getMutableState(env, tv.WorkflowID(), runID)
236
s.True(proto.Equal(config1, ms.State.ExecutionInfo.GetTimeSkippingInfo().GetConfig()))
237
events := collectOptionsEvents()
238
s.Len(events, 1)
239
s.True(proto.Equal(config1, events[0].GetWorkflowExecutionOptionsUpdatedEventAttributes().GetTimeSkippingConfig()))
240
241
// Second update: change the max_elapsed_duration duration.
242
config2 := &commonpb.TimeSkippingConfig{
243
Enabled: true,
244
FastForward: durationpb.New(2 * time.Hour),
245
}
246
updateOptions(config2)
247
248
ms = s.getMutableState(env, tv.WorkflowID(), runID)
249
s.True(proto.Equal(config2, ms.State.ExecutionInfo.GetTimeSkippingInfo().GetConfig()))
250
events = collectOptionsEvents()
251
s.Len(events, 2)
252
s.True(proto.Equal(config2, events[1].GetWorkflowExecutionOptionsUpdatedEventAttributes().GetTimeSkippingConfig()))
253
254
// Third update: disable time-skipping.
255
config3 := &commonpb.TimeSkippingConfig{Enabled: false}
256
updateOptions(config3)
257
258
ms = s.getMutableState(env, tv.WorkflowID(), runID)
259
s.True(proto.Equal(config3, ms.State.ExecutionInfo.GetTimeSkippingInfo().GetConfig()))
260
events = collectOptionsEvents()
261
s.Len(events, 3)
262
s.True(proto.Equal(config3, events[2].GetWorkflowExecutionOptionsUpdatedEventAttributes().GetTimeSkippingConfig()))
263
}
264
265
// TestTimeSkipping_ResetWithUpdateOptions verifies that resetting a workflow with a
266
// PostResetOperation that sets TimeSkippingConfig persists the config in the new run's
267
// mutable state and emits a WorkflowExecutionOptionsUpdated history event whose
268
// attributes carry the full config.
269
func (s *TimeSkippingTestSuite) TestTimeSkipping_ResetWithUpdateOptions() {
270
env := testcore.NewEnv(s.T())
271
env.OverrideDynamicConfig(dynamicconfig.TimeSkippingEnabled, true)
272
tv := testvars.New(s.T())
273
ctx := s.Context()
274
275
// Start a workflow and drain the first workflow task to establish a reset point.
276
startResp, err := env.FrontendClient().StartWorkflowExecution(ctx, &workflowservice.StartWorkflowExecutionRequest{
277
RequestId: uuid.NewString(),
278
Namespace: env.Namespace().String(),
279
WorkflowId: tv.WorkflowID(),
280
WorkflowType: tv.WorkflowType(),
281
TaskQueue: tv.TaskQueue(),
282
WorkflowRunTimeout: durationpb.New(100 * time.Second),
283
WorkflowTaskTimeout: durationpb.New(10 * time.Second),
284
})
285
s.NoError(err)
286
runID := startResp.RunId
287
288
_, err = env.TaskPoller().PollAndHandleWorkflowTask(tv, taskpoller.DrainWorkflowTask)
289
s.NoError(err)
290
291
// Find the WorkflowTaskCompleted event ID to use as the reset point.
292
histResp, err := env.FrontendClient().GetWorkflowExecutionHistory(ctx, &workflowservice.GetWorkflowExecutionHistoryRequest{
293
Namespace: env.Namespace().String(),
294
Execution: &commonpb.WorkflowExecution{WorkflowId: tv.WorkflowID(), RunId: runID},
295
})
296
s.NoError(err)
297
var wftCompletedEventID int64
298
for _, e := range histResp.History.Events {
299
if e.EventType == enumspb.EVENT_TYPE_WORKFLOW_TASK_COMPLETED {
300
wftCompletedEventID = e.EventId
301
break
302
}
303
}
304
s.NotZero(wftCompletedEventID)
305
306
// Reset with PostResetOperations that sets TimeSkippingConfig.
307
inputConfig := &commonpb.TimeSkippingConfig{
308
Enabled: true,
309
FastForward: durationpb.New(time.Hour)}
310
resetResp, err := env.FrontendClient().ResetWorkflowExecution(ctx, &workflowservice.ResetWorkflowExecutionRequest{
311
Namespace: env.Namespace().String(),
312
WorkflowExecution: &commonpb.WorkflowExecution{WorkflowId: tv.WorkflowID(), RunId: runID},
313
Reason: "test-timeskipping-reset",
314
RequestId: uuid.NewString(),
315
WorkflowTaskFinishEventId: wftCompletedEventID,
316
PostResetOperations: []*workflowpb.PostResetOperation{
317
{
318
Variant: &workflowpb.PostResetOperation_UpdateWorkflowOptions_{
319
UpdateWorkflowOptions: &workflowpb.PostResetOperation_UpdateWorkflowOptions{
320
WorkflowExecutionOptions: &workflowpb.WorkflowExecutionOptions{TimeSkippingConfig: inputConfig},
321
UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"time_skipping_config"}},
322
},
323
},
324
},
325
},
326
})
327
s.NoError(err)
328
newRunID := resetResp.RunId
329
330
// New run's mutable state must have the config.
331
ms := s.getMutableState(env, tv.WorkflowID(), newRunID)
332
s.True(proto.Equal(inputConfig, ms.State.ExecutionInfo.GetTimeSkippingInfo().GetConfig()))
333
334
// New run's history must contain a WorkflowExecutionOptionsUpdated event with the config.
335
histResp, err = env.FrontendClient().GetWorkflowExecutionHistory(ctx, &workflowservice.GetWorkflowExecutionHistoryRequest{
336
Namespace: env.Namespace().String(),
337
Execution: &commonpb.WorkflowExecution{WorkflowId: tv.WorkflowID(), RunId: newRunID},
338
})
339
s.NoError(err)
340
var optionsUpdatedEvent *historypb.HistoryEvent
341
for _, e := range histResp.History.Events {
342
if e.EventType == enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_OPTIONS_UPDATED {
343
optionsUpdatedEvent = e
344
break
345
}
346
}
347
s.NotNil(optionsUpdatedEvent, "expected WorkflowExecutionOptionsUpdated event in new run history")
348
s.True(proto.Equal(inputConfig, optionsUpdatedEvent.GetWorkflowExecutionOptionsUpdatedEventAttributes().GetTimeSkippingConfig()))
349
}
350
351
func (s *TimeSkippingTestSuite) getMutableState(env *testcore.TestEnv, workflowID, runID string) *persistence.GetWorkflowExecutionResponse {
352
shardID := common.WorkflowIDToHistoryShard(
353
env.NamespaceID().String(),
354
workflowID,
355
env.GetTestClusterConfig().HistoryConfig.NumHistoryShards,
356
)
357
ms, err := env.GetTestCluster().ExecutionManager().GetWorkflowExecution(s.Context(), &persistence.GetWorkflowExecutionRequest{
358
ShardID: shardID,
359
NamespaceID: env.NamespaceID().String(),
360
WorkflowID: workflowID,
361
RunID: runID,
362
ArchetypeID: chasm.WorkflowArchetypeID,
363
})
364
s.NoError(err)
365
return ms
366
}
367
368
// startWorkflowWithTimeSkipping starts a workflow with time-skipping enabled
369
// and a caller-specified run timeout. Used by tests that need the run timeout
370
// to be long enough to fit a virtual-time skip.
371
func (s *TimeSkippingTestSuite) startWorkflowWithTimeSkipping(env *testcore.TestEnv, tv *testvars.TestVars, runTimeout time.Duration) string {
372
resp, err := env.FrontendClient().StartWorkflowExecution(s.Context(), &workflowservice.StartWorkflowExecutionRequest{
373
RequestId: uuid.NewString(),
374
Namespace: env.Namespace().String(),
375
WorkflowId: tv.WorkflowID(),
376
WorkflowType: tv.WorkflowType(),
377
TaskQueue: tv.TaskQueue(),
378
WorkflowRunTimeout: durationpb.New(runTimeout),
379
WorkflowTaskTimeout: durationpb.New(10 * time.Second),
380
TimeSkippingConfig: &commonpb.TimeSkippingConfig{Enabled: true},
381
})
382
s.NoError(err)
383
return resp.GetRunId()
384
}
385
386
// scheduleActivityCmd returns a ScheduleActivityTask command that uses tv for all names /
387
// queue / timeout values.
388
func scheduleActivityCmd(tv *testvars.TestVars) *commandpb.Command {
389
return &commandpb.Command{
390
CommandType: enumspb.COMMAND_TYPE_SCHEDULE_ACTIVITY_TASK,
391
Attributes: &commandpb.Command_ScheduleActivityTaskCommandAttributes{
392
ScheduleActivityTaskCommandAttributes: &commandpb.ScheduleActivityTaskCommandAttributes{
393
ActivityId: tv.ActivityID(),
394
ActivityType: tv.ActivityType(),
395
TaskQueue: tv.TaskQueue(),
396
ScheduleToCloseTimeout: durationpb.New(30 * time.Second),
397
},
398
},
399
}
400
}
401
402
// startTimerCmd returns a StartTimer command with the given duration and timer ID.
403
func startTimerCmd(timerID string, d time.Duration) *commandpb.Command {
404
return &commandpb.Command{
405
CommandType: enumspb.COMMAND_TYPE_START_TIMER,
406
Attributes: &commandpb.Command_StartTimerCommandAttributes{
407
StartTimerCommandAttributes: &commandpb.StartTimerCommandAttributes{
408
TimerId: timerID,
409
StartToFireTimeout: durationpb.New(d),
410
},
411
},
412
}
413
}
414
415
// completeWorkflowCmd returns a CompleteWorkflowExecution command.
416
func completeWorkflowCmd() *commandpb.Command {
417
return &commandpb.Command{
418
CommandType: enumspb.COMMAND_TYPE_COMPLETE_WORKFLOW_EXECUTION,
419
Attributes: &commandpb.Command_CompleteWorkflowExecutionCommandAttributes{
420
CompleteWorkflowExecutionCommandAttributes: &commandpb.CompleteWorkflowExecutionCommandAttributes{},
421
},
422
}
423
}
424
425
// hasEventType returns true if any event in the slice has the given type.
426
func hasEventType(events []*historypb.HistoryEvent, t enumspb.EventType) bool {
427
for _, e := range events {
428
if e.GetEventType() == t {
429
return true
430
}
431
}
432
return false
433
}
434
435
// TestTimeSkipping_TimerAndActivity verifies that when a workflow has both a long user
436
// timer and a pending activity, time-skipping is blocked until the activity completes.
437
// Once the activity is done and the workflow task is drained, time-skipping fires and
438
// moves the timer's visibility timestamp to near-now, so the timer fires quickly.
439
//
440
// Sequence:
441
//
442
// WT1 → schedule activity + start 1-hour timer
443
// AT1 → complete activity
444
// WT2 → drain (return no commands; triggers time-skipping on close)
445
// WT3 → complete workflow (timer has fired)
446
func (s *TimeSkippingTestSuite) TestTimeSkipping_TimerAndActivity() {
447
env := testcore.NewEnv(s.T())
448
env.OverrideDynamicConfig(dynamicconfig.TimeSkippingEnabled, true)
449
tv := testvars.New(s.T())
450
451
// Run timeout must exceed the 1h timer; otherwise skip shifts the run-timeout
452
// task into the past and the workflow times out before WT3 can fire.
453
runID := s.startWorkflowWithTimeSkipping(env, tv, 2*time.Hour)
454
poller := taskpoller.New(s.T(), env.FrontendClient(), env.Namespace().String())
455
456
// WT 1: simultaneously schedule an activity and start a 1-hour timer.
457
// Time-skipping cannot fire while both are pending.
458
_, err := poller.PollAndHandleWorkflowTask(tv, func(_ *workflowservice.PollWorkflowTaskQueueResponse) (*workflowservice.RespondWorkflowTaskCompletedRequest, error) {
459
return &workflowservice.RespondWorkflowTaskCompletedRequest{
460
Commands: []*commandpb.Command{
461
scheduleActivityCmd(tv),
462
startTimerCmd("timer-1", time.Hour),
463
},
464
}, nil
465
})
466
s.NoError(err)
467
468
// Activity: complete it. After this, only the 1-hour timer is pending.
469
_, err = poller.PollAndHandleActivityTask(tv, taskpoller.CompleteActivityTask(tv))
470
s.NoError(err)
471
472
// WT 2: drain (return no commands). closeTransaction fires time-skipping here because
473
// the workflow is now idle with a pending timer → regenerates the timer task at near-now.
474
_, err = poller.PollAndHandleWorkflowTask(tv, taskpoller.DrainWorkflowTask)
475
s.NoError(err)
476
477
// WT 3: timer has fired (due to time-skipping); complete the workflow.
478
_, err = poller.PollAndHandleWorkflowTask(tv, func(_ *workflowservice.PollWorkflowTaskQueueResponse) (*workflowservice.RespondWorkflowTaskCompletedRequest, error) {
479
return &workflowservice.RespondWorkflowTaskCompletedRequest{
480
Commands: []*commandpb.Command{completeWorkflowCmd()},
481
}, nil
482
})
483
s.NoError(err)
484
485
// Verify history.
486
history := env.GetHistory(env.Namespace().String(), &commonpb.WorkflowExecution{WorkflowId: tv.WorkflowID(), RunId: runID})
487
s.True(hasEventType(history, enumspb.EVENT_TYPE_TIMER_FIRED), "timer must have fired via time-skipping")
488
s.True(hasEventType(history, enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_TIME_SKIPPING_TRANSITIONED),
489
"time-skipping transitioned event expected")
490
s.True(hasEventType(history, enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_COMPLETED), "workflow must complete")
491
}
492
493
// TestTimeSkipping_ActivityRetryBackoff verifies that when the only in-flight work
494
// is an activity waiting out a retry backoff, time-skipping fires: it advances virtual
495
// time to the activity's next-attempt time and re-stamps the activity retry timer to
496
// near-now wall, so the retry is dispatched promptly instead of after the full backoff.
497
//
498
// Sequence:
499
//
500
// WT1 → schedule an activity with a 1h retry InitialInterval (MaximumAttempts=2)
501
// AT1 → fail attempt 1; the server schedules the retry 1h out (virtual). The workflow
502
// is now idle except for the backoff activity, so the close transaction skips ~1h
503
// and re-stamps the ActivityRetryTimerTask to ~now wall.
504
// AT2 → retry is dispatchable promptly (well under 1h wall); complete it.
505
// WT2 → activity completed → complete the workflow.
506
func (s *TimeSkippingTestSuite) TestTimeSkipping_ActivityRetryBackoff() {
507
env := testcore.NewEnv(s.T())
508
env.OverrideDynamicConfig(dynamicconfig.TimeSkippingEnabled, true)
509
tv := testvars.New(s.T())
510
511
wallStart := time.Now()
512
// Run timeout must exceed the (virtual) backoff so the skipped-forward run-timeout
513
// task doesn't fire the workflow before the activity retries.
514
runID := s.startWorkflowWithTimeSkipping(env, tv, 4*time.Hour)
515
poller := taskpoller.New(s.T(), env.FrontendClient(), env.Namespace().String())
516
517
// WT1: schedule an activity that backs off 1h between attempts. ScheduleToClose
518
// must exceed the backoff so the retry isn't cut off by the schedule-to-close deadline.
519
_, err := poller.PollAndHandleWorkflowTask(tv, func(_ *workflowservice.PollWorkflowTaskQueueResponse) (*workflowservice.RespondWorkflowTaskCompletedRequest, error) {
520
return &workflowservice.RespondWorkflowTaskCompletedRequest{
521
Commands: []*commandpb.Command{
522
{
523
CommandType: enumspb.COMMAND_TYPE_SCHEDULE_ACTIVITY_TASK,
524
Attributes: &commandpb.Command_ScheduleActivityTaskCommandAttributes{
525
ScheduleActivityTaskCommandAttributes: &commandpb.ScheduleActivityTaskCommandAttributes{
526
ActivityId: tv.ActivityID(),
527
ActivityType: tv.ActivityType(),
528
TaskQueue: tv.TaskQueue(),
529
ScheduleToCloseTimeout: durationpb.New(3 * time.Hour),
530
StartToCloseTimeout: durationpb.New(30 * time.Second),
531
RetryPolicy: &commonpb.RetryPolicy{
532
InitialInterval: durationpb.New(time.Hour),
533
BackoffCoefficient: 1.0,
534
MaximumAttempts: 2,
535
},
536
},
537
},
538
},
539
},
540
}, nil
541
})
542
s.NoError(err)
543
544
// AT1: fail the activity, triggering a 1h retry backoff and (because the workflow
545
// is otherwise idle) a time-skipping transition on the close transaction.
546
_, err = poller.PollAndHandleActivityTask(tv, func(_ *workflowservice.PollActivityTaskQueueResponse) (*workflowservice.RespondActivityTaskCompletedRequest, error) {
547
return nil, errors.New("fail attempt 1")
548
})
549
s.NoError(err)
550
551
// AT2: the retry is dispatchable promptly thanks to time-skipping; complete it.
552
_, err = poller.PollAndHandleActivityTask(tv, taskpoller.CompleteActivityTask(tv))
553
s.NoError(err)
554
555
// WT2: activity completed → complete the workflow.
556
_, err = poller.PollAndHandleWorkflowTask(tv, func(_ *workflowservice.PollWorkflowTaskQueueResponse) (*workflowservice.RespondWorkflowTaskCompletedRequest, error) {
557
return &workflowservice.RespondWorkflowTaskCompletedRequest{
558
Commands: []*commandpb.Command{completeWorkflowCmd()},
559
}, nil
560
})
561
s.NoError(err)
562
wallElapsed := time.Since(wallStart)
563
564
history := env.GetHistory(env.Namespace().String(), &commonpb.WorkflowExecution{WorkflowId: tv.WorkflowID(), RunId: runID})
565
s.True(hasEventType(history, enumspb.EVENT_TYPE_ACTIVITY_TASK_COMPLETED),
566
"activity must complete on its retry attempt")
567
s.True(hasEventType(history, enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_TIME_SKIPPING_TRANSITIONED),
568
"time-skipping transitioned event expected (the activity retry backoff was skipped)")
569
s.True(hasEventType(history, enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_COMPLETED), "workflow must complete")
570
571
// Wall elapsed must be well under the 1h backoff — the retry was skipped, not waited out.
572
s.Less(wallElapsed, 3*time.Second,
573
"test wall elapsed = %v; the activity retry should be dispatched promptly after skipping the 1h backoff",
574
wallElapsed)
575
}
576
577
func (s *TimeSkippingTestSuite) TestTimeSkipping_PendingSignalExternalBlocksSkip() {
578
env := testcore.NewEnv(s.T())
579
env.OverrideDynamicConfig(dynamicconfig.TimeSkippingEnabled, true)
580
tv := testvars.New(s.T())
581
ctx := s.Context()
582
583
// Target workflow B. No worker polls B; the SignalExternal RPC will land a
584
// WorkflowExecutionSignaled event in B's history directly. Distinct task
585
// queue so B's idle first WT can't interfere with the SDK worker.
586
tvB := tv.WithWorkflowIDNumber(2).WithTaskQueueNumber(2)
587
_, err := env.FrontendClient().StartWorkflowExecution(ctx, &workflowservice.StartWorkflowExecutionRequest{
588
RequestId: uuid.NewString(),
589
Namespace: env.Namespace().String(),
590
WorkflowId: tvB.WorkflowID(),
591
WorkflowType: tvB.WorkflowType(),
592
TaskQueue: tvB.TaskQueue(),
593
WorkflowRunTimeout: durationpb.New(2 * time.Hour),
594
WorkflowTaskTimeout: durationpb.New(10 * time.Second),
595
})
596
s.NoError(err)
597
598
// CoordinatorWorkflow: emits the 1h timer and the SignalExternal command in
599
// the same WFT response, then waits for whichever future resolves first.
600
coordinatorWorkflow := func(wfCtx workflow.Context, targetWorkflowID string) error {
601
timerFuture := workflow.NewTimer(wfCtx, time.Hour)
602
signalFuture := workflow.SignalExternalWorkflow(
603
wfCtx, targetWorkflowID, "", "test-pending-signal", nil)
604
605
workflow.NewSelector(wfCtx).
606
AddFuture(timerFuture, func(_ workflow.Future) {}).
607
AddFuture(signalFuture, func(_ workflow.Future) {}).
608
Select(wfCtx)
609
return nil
610
}
611
const coordinatorTypeName = "CoordinatorWorkflow"
612
env.SdkWorker().RegisterWorkflowWithOptions(coordinatorWorkflow, workflow.RegisterOptions{
613
Name: coordinatorTypeName,
614
})
615
616
// SDK's StartWorkflowOptions doesn't expose TimeSkippingConfig (as of SDK
617
// v1.41), so start workflow A directly through the frontend. Use the SDK
618
// worker's task queue so the registered coordinator picks up the WT.
619
input, err := converter.GetDefaultDataConverter().ToPayloads(tvB.WorkflowID())
620
s.NoError(err)
621
622
tvA := tv.WithWorkflowIDNumber(1)
623
wallStart := time.Now()
624
aResp, err := env.FrontendClient().StartWorkflowExecution(ctx, &workflowservice.StartWorkflowExecutionRequest{
625
RequestId: uuid.NewString(),
626
Namespace: env.Namespace().String(),
627
WorkflowId: tvA.WorkflowID(),
628
WorkflowType: &commonpb.WorkflowType{Name: coordinatorTypeName},
629
TaskQueue: &taskqueuepb.TaskQueue{Name: env.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
630
Input: input,
631
WorkflowRunTimeout: durationpb.New(2 * time.Hour),
632
WorkflowTaskTimeout: durationpb.New(10 * time.Second),
633
TimeSkippingConfig: &commonpb.TimeSkippingConfig{Enabled: true},
634
})
635
s.NoError(err)
636
637
// Wait for A to finish through the SDK.
638
err = env.SdkClient().GetWorkflow(ctx, tvA.WorkflowID(), aResp.RunId).Get(ctx, nil)
639
s.NoError(err)
640
wallElapsed := time.Since(wallStart)
641
642
history := env.GetHistory(env.Namespace().String(), &commonpb.WorkflowExecution{
643
WorkflowId: tvA.WorkflowID(),
644
RunId: aResp.RunId,
645
})
646
647
// The signal future resolved (its completion event landed in A's history).
648
s.True(hasEventType(history, enumspb.EVENT_TYPE_EXTERNAL_WORKFLOW_EXECUTION_SIGNALED),
649
"ExternalWorkflowExecutionSignaled event must appear in A's history")
650
651
// The timer never fired — the signal won the Selector. If the new branch in
652
// hasInflightWorkToPreventTimeSkipping were missing, skip could fire at WT1
653
// close, shift the timer to near-now, and race the signal — making this
654
// assertion flaky.
655
s.False(hasEventType(history, enumspb.EVENT_TYPE_TIMER_FIRED),
656
"TimerFired must NOT appear — the signal future must resolve before the 1h timer fires")
657
658
// Workflow A closed via the signal branch.
659
s.True(hasEventType(history, enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_COMPLETED),
660
"workflow A must complete")
661
662
// Wall elapsed must be well under the 1h timer — the workflow should
663
// finish as soon as the signal completes (sub-second on a healthy cluster).
664
s.Less(wallElapsed, 5*time.Minute,
665
"test wall elapsed = %v; the workflow should complete promptly after the signal succeeds, well before the 1h timer would fire",
666
wallElapsed)
667
}
668
669
// TestTimeSkipping_StartWithDelay_NoBound verifies that time-skipping with no
670
// bound shifts a WorkflowStartDelay backoff into the near-now wall-clock window:
671
// the first WT becomes available immediately instead of waiting wallStart + 1h.
672
//
673
// Sequence:
674
//
675
// Start workflow with WorkflowStartDelay = 1h, TimeSkippingConfig{Enabled: true},
676
// no bound. On the close transaction of WorkflowExecutionStarted,
677
// calculateTimeSkippingTransition picks the backoff (only candidate;
678
// !HadOrHasWorkflowTask && ExecutionTime > StartTime), skips by 1h,
679
// accumulated = 1h. RegenerateTimerTasksForTimeSkipping step (4) re-emits the
680
// WorkflowBackoffTimerTask with VisibilityTimestamp ≈ wallStart
681
// (= virtual_executionTime − accumulated = (wallStart + 1h) − 1h).
682
// WT1 → complete workflow.
683
//
684
// Assertions:
685
//
686
// 1. WT1 polled in < 5min wall (would block 1h without skip, exceeding long-poll).
687
// 2. At least two WorkflowBackoffTimerTask writes (initial + regenerated).
688
// 3. All backoff tasks are typed WORKFLOW_BACKOFF_TYPE_DELAY_START
689
// (no cron, attempt == 1).
690
// 4. The latest task's VisibilityTimestamp is ≈ wallStart, not wallStart + 1h.
691
func (s *TimeSkippingTestSuite) TestTimeSkipping_StartWithDelay() {
692
env := testcore.NewEnv(
693
s.T(),
694
testcore.WithHistoryTaskRecorder(),
695
testcore.WithDynamicConfig(dynamicconfig.TimeSkippingEnabled, true),
696
)
697
tv := testvars.New(s.T())
698
699
const (
700
startDelay = time.Hour
701
shiftTol = 5 * time.Second
702
)
703
wallStart := time.Now()
704
705
startResp, err := env.FrontendClient().StartWorkflowExecution(s.Context(), &workflowservice.StartWorkflowExecutionRequest{
706
RequestId: uuid.NewString(),
707
Namespace: env.Namespace().String(),
708
WorkflowId: tv.WorkflowID(),
709
WorkflowType: tv.WorkflowType(),
710
TaskQueue: tv.TaskQueue(),
711
WorkflowRunTimeout: durationpb.New(24 * time.Hour),
712
WorkflowTaskTimeout: durationpb.New(10 * time.Second),
713
TimeSkippingConfig: &commonpb.TimeSkippingConfig{Enabled: true},
714
WorkflowStartDelay: durationpb.New(startDelay),
715
})
716
s.NoError(err)
717
runID := startResp.RunId
718
719
poller := taskpoller.New(s.T(), env.FrontendClient(), env.Namespace().String())
720
_, err = poller.PollAndHandleWorkflowTask(tv, func(_ *workflowservice.PollWorkflowTaskQueueResponse) (*workflowservice.RespondWorkflowTaskCompletedRequest, error) {
721
return &workflowservice.RespondWorkflowTaskCompletedRequest{
722
Commands: []*commandpb.Command{completeWorkflowCmd()},
723
}, nil
724
})
725
s.NoError(err)
726
727
elapsed := time.Since(wallStart)
728
s.Less(elapsed, shiftTol, "skip should have shifted the 1h start delay into near-now wall-clock; took %v", elapsed)
729
730
recorder := env.GetTestCluster().GetHistoryTaskRecorder()
731
s.NotNil(recorder)
732
recorded := recorder.GetRecordedTasksByCategoryFiltered(historytasks.CategoryTimer, testcore.TaskFilter{
733
NamespaceID: env.NamespaceID().String(),
734
WorkflowID: tv.WorkflowID(),
735
RunID: runID,
736
})
737
var backoffTasks []*historytasks.WorkflowBackoffTimerTask
738
for _, rec := range recorded {
739
if t, ok := rec.Task.(*historytasks.WorkflowBackoffTimerTask); ok {
740
backoffTasks = append(backoffTasks, t)
741
}
742
}
743
s.GreaterOrEqual(len(backoffTasks), 2, "expected initial + regenerated WorkflowBackoffTimerTask (two writes)")
744
for _, t := range backoffTasks {
745
s.Equal(enumsspb.WORKFLOW_BACKOFF_TYPE_DELAY_START, t.WorkflowBackoffType,
746
"all backoff tasks for start-with-delay must have type DELAY_START")
747
}
748
if len(backoffTasks) >= 1 {
749
latest := backoffTasks[len(backoffTasks)-1]
750
s.Less(latest.VisibilityTimestamp.Sub(wallStart), shiftTol,
751
"regenerated backoff task VisibilityTime must be ≈ wallStart (= virtual exec − accum), got %v vs wallStart %v",
752
latest.VisibilityTimestamp, wallStart)
753
}
754
}
755
756
// TestTimeSkipping_CanceledTimerNotUsedAsSkipTarget confirms that a timer
757
// canceled via command is excluded from the skip-target calculation.
758
// ApplyTimerCanceledEvent deletes the timer from pendingTimerInfoIDs, so it is
759
// invisible to calculateTimeSkippingTransition.
760
//
761
// If a canceled timer were mistakenly used, the accumulated skip would equal
762
// the sum of all timers' durations rather than just the surviving timers'.
763
//
764
// Sequence:
765
//
766
// WT1 → start timer-A (1h) + timer-B (5h)
767
// Skip to timer-A (nearest), accumulated = 1h, timer-A fires → WT2
768
// WT2 → cancel timer-B + start timer-C (2h)
769
// Skip to timer-C (2h), accumulated = 1h + 2h = 3h, timer-C fires → WT3
770
// WT3 → complete workflow
771
// Verify: AccumulatedSkippedDuration ≈ 3h (NOT 1h+4h=5h from canceled timer-B)
772
func (s *TimeSkippingTestSuite) TestTimeSkipping_CanceledTimerNotUsedAsSkipTarget() {
773
env := testcore.NewEnv(s.T())
774
env.OverrideDynamicConfig(dynamicconfig.TimeSkippingEnabled, true)
775
tv := testvars.New(s.T())
776
777
const (
778
timerADuration = time.Hour
779
timerBDuration = 5 * time.Hour
780
timerCDuration = 2 * time.Hour
781
runTimeout = 10 * time.Hour
782
)
783
784
runID := s.startWorkflowWithTimeSkipping(env, tv, runTimeout)
785
poller := taskpoller.New(s.T(), env.FrontendClient(), env.Namespace().String())
786
787
// WT1: start timer-A (1h) + timer-B (5h). No activity → time skipping fires
788
// on close-tx and targets timer-A (nearest candidate).
789
_, err := poller.PollAndHandleWorkflowTask(tv, func(_ *workflowservice.PollWorkflowTaskQueueResponse) (*workflowservice.RespondWorkflowTaskCompletedRequest, error) {
790
return &workflowservice.RespondWorkflowTaskCompletedRequest{
791
Commands: []*commandpb.Command{
792
startTimerCmd("timer-A", timerADuration),
793
startTimerCmd("timer-B", timerBDuration),
794
},
795
}, nil
796
})
797
s.NoError(err)
798
799
// WT2: timer-A has fired. Cancel timer-B and start timer-C (2h). After this
800
// WFT, only timer-C is a candidate — timer-B is gone from pendingTimerInfoIDs.
801
_, err = poller.PollAndHandleWorkflowTask(tv, func(_ *workflowservice.PollWorkflowTaskQueueResponse) (*workflowservice.RespondWorkflowTaskCompletedRequest, error) {
802
return &workflowservice.RespondWorkflowTaskCompletedRequest{
803
Commands: []*commandpb.Command{
804
{
805
CommandType: enumspb.COMMAND_TYPE_CANCEL_TIMER,
806
Attributes: &commandpb.Command_CancelTimerCommandAttributes{
807
CancelTimerCommandAttributes: &commandpb.CancelTimerCommandAttributes{
808
TimerId: "timer-B",
809
},
810
},
811
},
812
startTimerCmd("timer-C", timerCDuration),
813
},
814
}, nil
815
})
816
s.NoError(err)
817
818
// WT3: timer-C fired (skip targeted it at 2h, not the 4h remaining on
819
// canceled timer-B). Complete workflow.
820
_, err = poller.PollAndHandleWorkflowTask(tv, func(_ *workflowservice.PollWorkflowTaskQueueResponse) (*workflowservice.RespondWorkflowTaskCompletedRequest, error) {
821
return &workflowservice.RespondWorkflowTaskCompletedRequest{
822
Commands: []*commandpb.Command{completeWorkflowCmd()},
823
}, nil
824
})
825
s.NoError(err)
826
827
// Accumulated skip ≈ 1h + 2h = 3h. 1s tolerance covers real-time slack
828
// between skip transition and timer-fired and is far below the 2h margin
829
// between the right answer (3h) and the wrong answer (5h with canceled
830
// timer-B included).
831
ms := s.getMutableState(env, tv.WorkflowID(), runID)
832
accumulated := ms.State.ExecutionInfo.GetTimeSkippingInfo().GetAccumulatedSkippedDuration().AsDuration()
833
s.InDelta(float64(timerADuration+timerCDuration), float64(accumulated), float64(time.Second),
834
"AccumulatedSkippedDuration must equal sum of non-canceled timer durations (1h+2h=3h), not include the canceled timer-B")
835
836
history := env.GetHistory(env.Namespace().String(), &commonpb.WorkflowExecution{
837
WorkflowId: tv.WorkflowID(), RunId: runID,
838
})
839
s.True(hasEventType(history, enumspb.EVENT_TYPE_TIMER_CANCELED), "timer-B must be canceled")
840
s.True(hasEventType(history, enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_COMPLETED))
841
}
842
843
// TestWorkflowLifecycle_VirtualTimeContract is an end-to-end regression test
844
// that exercises the virtual-time contract across a full workflow lifecycle:
845
// activity execution before and after skip, skip transition, workflow close,
846
// close-time semantics, run-timeout task regeneration, activity-timeout task
847
// shifting, and retention task scheduling.
848
//
849
// Scenario:
850
//
851
// Start workflow with runTimeout = 4h, time-skipping enabled, namespace
852
// retention = 1 day.
853
// WT1 → schedule activity-1 + 1h user timer (activity blocks skip).
854
// AT1 → complete activity-1 (virtual time barely advances — activity is fast).
855
// WT2 → drain → closeTransaction fires skip transition (accumulated ≈ 1h).
856
// WT3 → timer-1 has fired; schedule activity-2 with 5min ScheduleToClose
857
// (this is when ActivityTimeoutTask for activity-2 is written, with
858
// virtual ScheduledTime ≈ wallStart + 1h).
859
// AT2 → complete activity-2.
860
// WT4 → complete workflow.
861
//
862
// Assertions:
863
//
864
// 1. Activity-1 events: ActivityTaskCompleted for activity-1 appears in history.
865
// 2. Skip: history has WorkflowExecutionTimeSkippingTransitioned event.
866
// 3. Workflow close: history has WorkflowExecutionCompleted.
867
// 4. Describe semantics:
868
// 4a. StartTime ≈ wallBeforeStart (wall frame, admission anchor).
869
// 4b. ExecutionTime == StartTime (no backoff configured).
870
// 4c. CloseTime − StartTime ≈ skip (virtual frame).
871
// 4d. CloseTime − ExecutionTime ≈ skip (reported duration).
872
// 5. Run-timeout task regenerated: two WorkflowRunTimeoutTask writes; the
873
// second has VisibilityTimestamp ≈ wallStart + runTimeout − skip (earlier
874
// by ≈1h than the first).
875
// 6. Retention task: DeleteHistoryEventTask has VisibilityTimestamp ≈
876
// wallClose + retention. The retention fires at real wall time
877
// "retention after close," not virtual time "retention after virtual close."
878
// Concretely: virtual deleteTime = virtualCloseTime + retention; after
879
// toRealTime this is (wallClose + skip) + retention − skip =
880
// wallClose + retention.
881
// 7. Activity-2 events: ActivityTaskScheduled for activity-2 has EventTime in
882
// virtual frame (EventTime − workflow StartTime ≥ ~skip). Completed event
883
// for activity-2 also appears.
884
// 8. Activity-2 timeout task: the ActivityTimeoutTask written when activity-2
885
// is scheduled has VisibilityTimestamp ≈ wallAtActivity2Schedule +
886
// scheduleToClose. Principle 2: even though virtual time is ~1h ahead when
887
// the activity is scheduled, the wall-clock VisibilityTimestamp must be
888
// anchored to real wall time (test allows ±3min tolerance).
889
//
890
// This is the single most comprehensive e2e test for the virtual-time system.
891
// If any of the four principles regresses, at least one of assertions 4–8 will
892
// fail.
893
func (s *TimeSkippingTestSuite) TestWorkflowLifecycle_VirtualTimeContract() {
894
env := testcore.NewEnv(
895
s.T(),
896
testcore.WithHistoryTaskRecorder(),
897
testcore.WithDynamicConfig(dynamicconfig.TimeSkippingEnabled, true),
898
)
899
tv := testvars.New(s.T())
900
901
const (
902
runTimeout = 4 * time.Hour
903
timerDuration = 1 * time.Hour // determines the amount of skip
904
// Namespace retention is 1 day per testcore.NewEnv (see test_env.go:165).
905
// Any future change to that default will require updating this constant.
906
namespaceRetention = 24 * time.Hour
907
// Margin for timing-related assertions: scheduling jitter between our
908
// wall-time samples and the server's, plus any clock drift. Must be
909
// less than timerDuration so the shift assertion differentiates the
910
// virtual-frame and wall-frame cases.
911
assertionMargin = 5 * time.Minute
912
// Second activity (scheduled AFTER skip) uses a 5-minute
913
// ScheduleToClose timeout. With accumulated skip = timerDuration (~1h),
914
// virtualToRealTime at the task-generator boundary should produce an
915
// ActivityTimeoutTask VisibilityTimestamp ≈ wallAtActivity2Schedule +
916
// activity2ScheduleToClose. Delta within activity2TimerMargin is
917
// allowed.
918
activity2ScheduleToClose = 5 * time.Minute
919
activity2TimerMargin = 3 * time.Minute
920
)
921
922
wallBeforeStart := time.Now()
923
runID := s.startWorkflowWithTimeSkipping(env, tv, runTimeout)
924
poller := taskpoller.New(s.T(), env.FrontendClient(), env.Namespace().String())
925
926
// WT1: schedule activity-1 + 1h user timer. Activity blocks skip.
927
tvActivity1 := tv.WithActivityIDNumber(1)
928
_, err := poller.PollAndHandleWorkflowTask(tv, func(_ *workflowservice.PollWorkflowTaskQueueResponse) (*workflowservice.RespondWorkflowTaskCompletedRequest, error) {
929
return &workflowservice.RespondWorkflowTaskCompletedRequest{
930
Commands: []*commandpb.Command{
931
scheduleActivityCmd(tvActivity1),
932
startTimerCmd("lifecycle-timer", timerDuration),
933
},
934
}, nil
935
})
936
s.NoError(err)
937
938
// AT1: complete activity-1. After this the workflow has only the pending
939
// user timer → skip becomes eligible.
940
_, err = poller.PollAndHandleActivityTask(tvActivity1, taskpoller.CompleteActivityTask(tv))
941
s.NoError(err)
942
943
// WT2: drain → closeTransaction fires the skip transition.
944
_, err = poller.PollAndHandleWorkflowTask(tv, taskpoller.DrainWorkflowTask)
945
s.NoError(err)
946
947
// WT3: timer-1 has fired via skip. Schedule activity-2 with a 5-minute
948
// ScheduleToClose timeout. Record the wall time so we can verify that the
949
// ActivityTimeoutTask's VisibilityTimestamp is anchored to wall clock
950
// (principle 2) and NOT to virtual time (which would put it ~1h+5min out).
951
wallAtActivity2Schedule := time.Now()
952
tvActivity2 := tv.WithActivityIDNumber(2)
953
_, err = poller.PollAndHandleWorkflowTask(tv, func(_ *workflowservice.PollWorkflowTaskQueueResponse) (*workflowservice.RespondWorkflowTaskCompletedRequest, error) {
954
return &workflowservice.RespondWorkflowTaskCompletedRequest{
955
Commands: []*commandpb.Command{
956
{
957
CommandType: enumspb.COMMAND_TYPE_SCHEDULE_ACTIVITY_TASK,
958
Attributes: &commandpb.Command_ScheduleActivityTaskCommandAttributes{
959
ScheduleActivityTaskCommandAttributes: &commandpb.ScheduleActivityTaskCommandAttributes{
960
ActivityId: tvActivity2.ActivityID(),
961
ActivityType: tvActivity2.ActivityType(),
962
TaskQueue: tv.TaskQueue(),
963
ScheduleToCloseTimeout: durationpb.New(activity2ScheduleToClose),
964
},
965
},
966
},
967
},
968
}, nil
969
})
970
s.NoError(err)
971
972
// AT2: complete activity-2.
973
_, err = poller.PollAndHandleActivityTask(tvActivity2, taskpoller.CompleteActivityTask(tv))
974
s.NoError(err)
975
976
// WT4: complete workflow.
977
_, err = poller.PollAndHandleWorkflowTask(tv, func(_ *workflowservice.PollWorkflowTaskQueueResponse) (*workflowservice.RespondWorkflowTaskCompletedRequest, error) {
978
return &workflowservice.RespondWorkflowTaskCompletedRequest{
979
Commands: []*commandpb.Command{completeWorkflowCmd()},
980
}, nil
981
})
982
s.NoError(err)
983
wallAfterClose := time.Now()
984
985
// ── Assertion 1/2/3: the three history events are present. ────────────────
986
history := env.GetHistory(env.Namespace().String(), &commonpb.WorkflowExecution{WorkflowId: tv.WorkflowID(), RunId: runID})
987
s.True(hasEventType(history, enumspb.EVENT_TYPE_ACTIVITY_TASK_COMPLETED), "activity must have completed")
988
s.True(hasEventType(history, enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_TIME_SKIPPING_TRANSITIONED), "skip transition must have happened")
989
s.True(hasEventType(history, enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_COMPLETED), "workflow must have completed")
990
991
// ── Assertion 4: StartTime, ExecutionTime, CloseTime have the right frames. ─
992
//
993
// StartTime = caller-provided admission time → WALL clock,
994
// ≈ wallBeforeStart (± scheduling jitter).
995
// ExecutionTime = StartTime + FirstWorkflowTaskBackoff.
996
// With no cron / WorkflowStartDelay / ContinueAsNew-backoff,
997
// FirstWorkflowTaskBackoff = 0 → ExecutionTime == StartTime.
998
// CloseTime = VIRTUAL time at WorkflowExecutionCompleted event.
999
// = wallClose + accumulatedSkip ≈ wallStart + skip.
1000
//
1001
// Consequences:
1002
// - CloseTime − StartTime ≈ skip (≈ timerDuration). (this is virtualDuration)
1003
// - CloseTime − ExecutionTime ≈ skip. (public reported duration)
1004
desc, err := env.FrontendClient().DescribeWorkflowExecution(s.Context(), &workflowservice.DescribeWorkflowExecutionRequest{
1005
Namespace: env.Namespace().String(),
1006
Execution: &commonpb.WorkflowExecution{WorkflowId: tv.WorkflowID(), RunId: runID},
1007
})
1008
s.NoError(err)
1009
execInfo := desc.GetWorkflowExecutionInfo()
1010
s.Equal(enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED, execInfo.GetStatus())
1011
s.NotNil(execInfo.GetStartTime(), "StartTime must be set")
1012
s.NotNil(execInfo.GetExecutionTime(), "ExecutionTime must be set")
1013
s.NotNil(execInfo.GetCloseTime(), "CloseTime must be set")
1014
1015
// 4a. StartTime ≈ wallBeforeStart (wall frame, admission anchor).
1016
startTime := execInfo.GetStartTime().AsTime()
1017
s.GreaterOrEqual(startTime, wallBeforeStart.Add(-assertionMargin),
1018
"StartTime %v is earlier than wallBeforeStart %v − margin; StartTime should be wall admission time",
1019
startTime, wallBeforeStart)
1020
s.LessOrEqual(startTime, wallBeforeStart.Add(assertionMargin),
1021
"StartTime %v is later than wallBeforeStart %v + margin; StartTime should be ≈ wall admission moment, NOT virtual-shifted",
1022
startTime, wallBeforeStart)
1023
1024
// 4b. ExecutionTime == StartTime (no backoff configured).
1025
executionTime := execInfo.GetExecutionTime().AsTime()
1026
s.Equal(startTime, executionTime,
1027
"ExecutionTime %v should equal StartTime %v because FirstWorkflowTaskBackoff = 0",
1028
executionTime, startTime)
1029
1030
// 4c. CloseTime − StartTime ≈ skip (virtual frame).
1031
virtualDuration := execInfo.GetCloseTime().AsTime().Sub(startTime)
1032
s.GreaterOrEqual(
1033
virtualDuration, timerDuration-assertionMargin,
1034
"Describe.CloseTime − Describe.StartTime = %v is shorter than the accumulated skip (~%v); CloseTime may not be written in virtual frame",
1035
virtualDuration, timerDuration,
1036
)
1037
// Upper bound: skip + actual wall time spent (very small in this test).
1038
// assertionMargin (5min) is comfortably above any realistic wall elapsed.
1039
s.LessOrEqual(
1040
virtualDuration, timerDuration+assertionMargin,
1041
"Describe.CloseTime − Describe.StartTime = %v exceeds accumulated skip + test wall time; something is wrong with the close-time computation",
1042
virtualDuration,
1043
)
1044
1045
// 4d. CloseTime − ExecutionTime ≈ skip (the "reported duration").
1046
reportedDuration := execInfo.GetCloseTime().AsTime().Sub(executionTime)
1047
s.GreaterOrEqual(
1048
reportedDuration, timerDuration-assertionMargin,
1049
"Reported duration (CloseTime − ExecutionTime) = %v is shorter than accumulated skip (~%v)",
1050
reportedDuration, timerDuration,
1051
)
1052
s.LessOrEqual(
1053
reportedDuration, timerDuration+assertionMargin,
1054
"Reported duration (CloseTime − ExecutionTime) = %v exceeds accumulated skip + test wall time",
1055
reportedDuration,
1056
)
1057
1058
// ── Assertion 5: WorkflowRunTimeoutTask regenerated with shifted timestamp. ─
1059
recorder := env.GetTestCluster().GetHistoryTaskRecorder()
1060
s.NotNil(recorder)
1061
recorded := recorder.GetRecordedTasksByCategoryFiltered(historytasks.CategoryTimer, testcore.TaskFilter{
1062
NamespaceID: env.NamespaceID().String(),
1063
WorkflowID: tv.WorkflowID(),
1064
RunID: runID,
1065
})
1066
1067
var runTimeoutTasks []*historytasks.WorkflowRunTimeoutTask
1068
var deleteHistoryTasks []*historytasks.DeleteHistoryEventTask
1069
var activityTimeoutTasks []*historytasks.ActivityTimeoutTask
1070
for _, rec := range recorded {
1071
switch t := rec.Task.(type) {
1072
case *historytasks.WorkflowRunTimeoutTask:
1073
runTimeoutTasks = append(runTimeoutTasks, t)
1074
case *historytasks.DeleteHistoryEventTask:
1075
deleteHistoryTasks = append(deleteHistoryTasks, t)
1076
case *historytasks.ActivityTimeoutTask:
1077
activityTimeoutTasks = append(activityTimeoutTasks, t)
1078
default:
1079
// other task types are not relevant to this assertion
1080
}
1081
}
1082
1083
s.GreaterOrEqual(
1084
len(runTimeoutTasks), 2,
1085
"expected initial + regenerated WorkflowRunTimeoutTask (two writes)",
1086
)
1087
1088
// Initial task at workflow start: VisibilityTimestamp ≈ wallStart + runTimeout.
1089
firstRTT := runTimeoutTasks[0]
1090
initialExpected := wallBeforeStart.Add(runTimeout)
1091
s.GreaterOrEqual(firstRTT.VisibilityTimestamp, initialExpected.Add(-assertionMargin),
1092
"initial WorkflowRunTimeoutTask %v is earlier than expected %v",
1093
firstRTT.VisibilityTimestamp, initialExpected)
1094
s.LessOrEqual(firstRTT.VisibilityTimestamp, initialExpected.Add(assertionMargin),
1095
"initial WorkflowRunTimeoutTask %v is later than expected %v",
1096
firstRTT.VisibilityTimestamp, initialExpected)
1097
1098
// Regenerated task after skip: VisibilityTimestamp ≈ wallStart + runTimeout − skip.
1099
latestRTT := runTimeoutTasks[len(runTimeoutTasks)-1]
1100
shiftedExpected := wallBeforeStart.Add(runTimeout).Add(-timerDuration)
1101
s.GreaterOrEqual(latestRTT.VisibilityTimestamp, shiftedExpected.Add(-assertionMargin),
1102
"regenerated WorkflowRunTimeoutTask %v is earlier than expected %v",
1103
latestRTT.VisibilityTimestamp, shiftedExpected)
1104
s.LessOrEqual(latestRTT.VisibilityTimestamp, shiftedExpected.Add(assertionMargin),
1105
"regenerated WorkflowRunTimeoutTask %v is later than expected %v (principle 3: skip must shift outstanding tasks earlier by accumulated duration)",
1106
latestRTT.VisibilityTimestamp, shiftedExpected)
1107
1108
// ── Assertion 6: DeleteHistoryEventTask fires at wallClose + retention. ───
1109
//
1110
// virtual deleteTime = virtualCloseTime + retention + tiny-jitter
1111
// = (wallClose + skip) + retention + tiny-jitter
1112
// VisibilityTimestamp = toRealTime(deleteTime)
1113
// = deleteTime − skip
1114
// = wallClose + retention + tiny-jitter
1115
//
1116
// RetentionTimerJitterDuration is overridden to 1s in functional tests
1117
// (testcore/dynamic_config_overrides.go:54), so the jitter is negligible.
1118
s.GreaterOrEqual(len(deleteHistoryTasks), 1, "expected a DeleteHistoryEventTask for retention")
1119
deleteTask := deleteHistoryTasks[len(deleteHistoryTasks)-1]
1120
retentionExpected := wallAfterClose.Add(namespaceRetention)
1121
s.GreaterOrEqual(deleteTask.VisibilityTimestamp, retentionExpected.Add(-assertionMargin),
1122
"DeleteHistoryEventTask VisibilityTimestamp %v is earlier than expected ~wallClose+retention %v (principle 2: retention should anchor on wall close, not virtual close)",
1123
deleteTask.VisibilityTimestamp, retentionExpected)
1124
s.LessOrEqual(deleteTask.VisibilityTimestamp, retentionExpected.Add(assertionMargin),
1125
"DeleteHistoryEventTask VisibilityTimestamp %v is later than expected ~wallClose+retention %v",
1126
deleteTask.VisibilityTimestamp, retentionExpected)
1127
1128
// ── Assertion 7: activity-2 events are stamped with virtual time. ─────────
1129
//
1130
// activity-2 was scheduled AFTER the skip fired, so by the time WT3's
1131
// closeTransaction runs, ms.timeSource is ~1h ahead of wall clock. The
1132
// ActivityTaskScheduled event's EventTime is stamped from hBuilder which
1133
// uses ms.timeSource, so it must be in virtual frame — specifically at
1134
// least (workflow StartTime + timerDuration − margin) into the "virtual
1135
// future." A wall-frame write would produce EventTime ≈ wallStart +
1136
// small_delta, which would be ~1h earlier than the expected virtual
1137
// timestamp.
1138
var activity2ScheduledEvent, activity2CompletedEvent *historypb.HistoryEvent
1139
for _, e := range history {
1140
switch e.GetEventType() {
1141
case enumspb.EVENT_TYPE_ACTIVITY_TASK_SCHEDULED:
1142
if attrs := e.GetActivityTaskScheduledEventAttributes(); attrs != nil && attrs.GetActivityId() == tvActivity2.ActivityID() {
1143
activity2ScheduledEvent = e
1144
}
1145
case enumspb.EVENT_TYPE_ACTIVITY_TASK_COMPLETED:
1146
// Completion event doesn't carry ActivityId directly; match by
1147
// ScheduledEventId pointing at activity-2's Scheduled event.
1148
if activity2ScheduledEvent != nil &&
1149
e.GetActivityTaskCompletedEventAttributes().GetScheduledEventId() == activity2ScheduledEvent.GetEventId() {
1150
activity2CompletedEvent = e
1151
}
1152
default:
1153
// other event types are not relevant to this assertion
1154
}
1155
}
1156
s.NotNil(activity2ScheduledEvent, "ActivityTaskScheduled event for activity-2 must exist")
1157
s.NotNil(activity2CompletedEvent, "ActivityTaskCompleted event for activity-2 must exist")
1158
1159
virtualScheduledDelta := activity2ScheduledEvent.GetEventTime().AsTime().Sub(execInfo.GetStartTime().AsTime())
1160
s.GreaterOrEqual(
1161
virtualScheduledDelta, timerDuration-assertionMargin,
1162
"activity-2 ActivityTaskScheduled EventTime − StartTime = %v is shorter than accumulated skip (~%v); event may not be stamped with virtual time",
1163
virtualScheduledDelta, timerDuration,
1164
)
1165
1166
// ── Assertion 8: activity-2 timeout task is anchored to wall clock. ──────
1167
//
1168
// When activity-2 is scheduled (WT3 close), timer_sequence.CreateNextActivityTimer
1169
// writes an ActivityTimeoutTask with:
1170
// VisibilityTimestamp = toRealTime(virtualScheduledTime + activity2ScheduleToClose)
1171
// = (wallAtActivity2Schedule + skip + timeout) − skip
1172
// = wallAtActivity2Schedule + activity2ScheduleToClose
1173
//
1174
// Principle 2: regardless of virtual time's offset, the wall-clock
1175
// VisibilityTimestamp must be ≈ wallAtActivity2Schedule + 5min. The test
1176
// tolerates ±3min drift (covers scheduling jitter between our wall-time
1177
// sample and the server's write, plus any internal processing delay).
1178
s.GreaterOrEqual(len(activityTimeoutTasks), 1, "expected at least one ActivityTimeoutTask written for activity-2")
1179
// Scan all recorded ActivityTimeoutTasks; find the one referencing activity-2's
1180
// ScheduledEventId. Each activity may generate multiple timeout tasks
1181
// (ScheduleToStart, ScheduleToClose, etc.), but in this test only
1182
// ScheduleToClose was configured.
1183
var activity2TimeoutTask *historytasks.ActivityTimeoutTask
1184
for _, t := range activityTimeoutTasks {
1185
if t.EventID == activity2ScheduledEvent.GetEventId() {
1186
activity2TimeoutTask = t
1187
break
1188
}
1189
}
1190
s.NotNil(activity2TimeoutTask, "expected an ActivityTimeoutTask for activity-2 (EventID=%d)", activity2ScheduledEvent.GetEventId())
1191
1192
activity2TimerExpected := wallAtActivity2Schedule.Add(activity2ScheduleToClose)
1193
s.False(
1194
activity2TimeoutTask.VisibilityTimestamp.Before(activity2TimerExpected.Add(-activity2TimerMargin)),
1195
"activity-2 ActivityTimeoutTask VisibilityTimestamp %v is earlier than expected ~wallAtSchedule+5min %v (principle 2: task must anchor to wall clock, not virtual)",
1196
activity2TimeoutTask.VisibilityTimestamp, activity2TimerExpected,
1197
)
1198
s.False(
1199
activity2TimeoutTask.VisibilityTimestamp.After(activity2TimerExpected.Add(activity2TimerMargin)),
1200
"activity-2 ActivityTimeoutTask VisibilityTimestamp %v is later than expected ~wallAtSchedule+5min %v; this would happen if the virtual ScheduledTime leaked into VisibilityTimestamp (would show ≈ wallAtSchedule + 1h + 5min)",
1201
activity2TimeoutTask.VisibilityTimestamp, activity2TimerExpected,
1202
)
1203
}
1204
1205
func (s *TimeSkippingFastForwardFunctionalSuite) TestTimeSkipping_ExecutionTimeoutTimesOutIdleWorkflow() {
1206
env := testcore.NewEnv(s.T())
1207
env.OverrideDynamicConfig(dynamicconfig.TimeSkippingEnabled, true)
1208
ctx := s.Context()
1209
1210
const executionTimeout = 5 * time.Minute
1211
1212
env.SdkWorker().RegisterWorkflowWithOptions(func(ctx workflow.Context) error {
1213
return workflow.Await(ctx, func() bool { return false })
1214
}, workflow.RegisterOptions{Name: "blockingConditionWorkflow"})
1215
1216
startWall := time.Now()
1217
workflowID := uuid.NewString()
1218
startResp, err := env.FrontendClient().StartWorkflowExecution(ctx, &workflowservice.StartWorkflowExecutionRequest{
1219
RequestId: uuid.NewString(),
1220
Namespace: env.Namespace().String(),
1221
WorkflowId: workflowID,
1222
WorkflowType: &commonpb.WorkflowType{Name: "blockingConditionWorkflow"},
1223
TaskQueue: &taskqueuepb.TaskQueue{Name: env.WorkerTaskQueue()},
1224
WorkflowExecutionTimeout: durationpb.New(executionTimeout),
1225
WorkflowTaskTimeout: durationpb.New(10 * time.Second),
1226
TimeSkippingConfig: &commonpb.TimeSkippingConfig{Enabled: true},
1227
})
1228
s.NoError(err)
1229
runID := startResp.RunId
1230
1231
// The workflow is idle (blocked on Await) with the execution timeout as its only skip
1232
// target, so it skips straight to the timeout and times out — without waiting out 5 min
1233
// of real time.
1234
run := env.SdkClient().GetWorkflow(ctx, workflowID, runID)
1235
err = run.Get(ctx, nil)
1236
var timeoutErr *sdktemporal.TimeoutError
1237
s.ErrorAs(err, &timeoutErr, "expected TimeoutError, got: %v", err)
1238
1239
s.Less(time.Since(startWall), executionTimeout,
1240
"time skipping must reach the execution timeout in far less than the 5 min real timeout")
1241
1242
hist := env.GetHistory(env.Namespace().String(), &commonpb.WorkflowExecution{WorkflowId: workflowID, RunId: runID})
1243
transitions := s.findTransitionedEvents(hist)
1244
s.Len(transitions, 1, "exactly one transition: skip to the execution timeout")
1245
s.False(transitions[0].GetWorkflowExecutionTimeSkippingTransitionedEventAttributes().GetDisabledAfterFastForward(),
1246
"timing out at the execution timeout is not a fast-forward disable")
1247
s.True(hasEventType(hist, enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_TIMED_OUT))
1248
}
1249
1250
func (s *TimeSkippingTestSuite) TestTimeSkippingTransitionEventOrdersAfterOptionsUpdated() {
1251
env := testcore.NewEnv(s.T())
1252
env.OverrideDynamicConfig(dynamicconfig.TimeSkippingEnabled, true)
1253
tv := testvars.New(s.T())
1254
ctx := s.Context()
1255
1256
// Start WITHOUT time skipping.
1257
startResp, err := env.FrontendClient().StartWorkflowExecution(ctx, &workflowservice.StartWorkflowExecutionRequest{
1258
RequestId: uuid.NewString(),
1259
Namespace: env.Namespace().String(),
1260
WorkflowId: tv.WorkflowID(),
1261
WorkflowType: tv.WorkflowType(),
1262
TaskQueue: tv.TaskQueue(),
1263
WorkflowRunTimeout: durationpb.New(24 * time.Hour),
1264
WorkflowTaskTimeout: durationpb.New(10 * time.Second),
1265
})
1266
s.NoError(err)
1267
runID := startResp.RunId
1268
1269
// WT1: schedule a user timer. Time skipping is off, so the workflow just goes idle with a
1270
// pending timer (no skip yet).
1271
_, err = env.TaskPoller().PollAndHandleWorkflowTask(tv, func(_ *workflowservice.PollWorkflowTaskQueueResponse) (*workflowservice.RespondWorkflowTaskCompletedRequest, error) {
1272
return &workflowservice.RespondWorkflowTaskCompletedRequest{
1273
Commands: []*commandpb.Command{startTimerCmd("t1", time.Hour)},
1274
}, nil
1275
})
1276
s.NoError(err)
1277
1278
// Enable time skipping. The workflow is idle with a pending timer, so this update's close-tx
1279
// emits a transition skipping to the timer — in the same transaction as OPTIONS_UPDATED.
1280
_, err = env.FrontendClient().UpdateWorkflowExecutionOptions(ctx, &workflowservice.UpdateWorkflowExecutionOptionsRequest{
1281
Namespace: env.Namespace().String(),
1282
WorkflowExecution: &commonpb.WorkflowExecution{WorkflowId: tv.WorkflowID(), RunId: runID},
1283
WorkflowExecutionOptions: &workflowpb.WorkflowExecutionOptions{
1284
TimeSkippingConfig: &commonpb.TimeSkippingConfig{Enabled: true},
1285
},
1286
UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"time_skipping_config"}},
1287
})
1288
s.NoError(err)
1289
1290
hist := env.GetHistory(env.Namespace().String(), &commonpb.WorkflowExecution{WorkflowId: tv.WorkflowID(), RunId: runID})
1291
1292
// The transition and the OPTIONS_UPDATED that triggered it must both be present, and the
1293
// OPTIONS_UPDATED must come first (the transition is the last event of the transaction).
1294
optionsUpdatedIdx, transitionIdx := -1, -1
1295
for i, e := range hist {
1296
switch e.GetEventType() {
1297
case enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_OPTIONS_UPDATED:
1298
optionsUpdatedIdx = i
1299
case enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_TIME_SKIPPING_TRANSITIONED:
1300
transitionIdx = i
1301
default:
1302
// other event types are irrelevant to this ordering check
1303
}
1304
}
1305
s.NotEqual(-1, optionsUpdatedIdx, "expected an OPTIONS_UPDATED event")
1306
s.NotEqual(-1, transitionIdx, "expected a TIME_SKIPPING_TRANSITIONED event")
1307
s.Less(optionsUpdatedIdx, transitionIdx,
1308
"OPTIONS_UPDATED (event %d) must precede the transition it triggered (event %d)",
1309
hist[optionsUpdatedIdx].GetEventId(), hist[transitionIdx].GetEventId())
1310
_, _ = env.FrontendClient().TerminateWorkflowExecution(ctx, &workflowservice.TerminateWorkflowExecutionRequest{
1311
Namespace: env.Namespace().String(),
1312
WorkflowExecution: &commonpb.WorkflowExecution{WorkflowId: tv.WorkflowID(), RunId: runID},
1313
Reason: "test cleanup",
1314
})
1315
}