go.temporal.io/server/tests/versioning_test.go
5157 LOC · 0 covered · 5157 uncovered · 0 ranges · 0 concepts · 0 introducers · 0 tests
1
// nolint:revive
2
package tests
3
4
import (
5
"errors"
6
"fmt"
7
"math/rand"
8
"strconv"
9
"strings"
10
"sync/atomic"
11
"testing"
12
"time"
13
14
"github.com/dgryski/go-farm"
15
commandpb "go.temporal.io/api/command/v1"
16
commonpb "go.temporal.io/api/common/v1"
17
enumspb "go.temporal.io/api/enums/v1"
18
"go.temporal.io/api/serviceerror"
19
taskqueuepb "go.temporal.io/api/taskqueue/v1"
20
"go.temporal.io/api/workflowservice/v1"
21
"go.temporal.io/sdk/activity"
22
sdkclient "go.temporal.io/sdk/client"
23
"go.temporal.io/sdk/temporal"
24
"go.temporal.io/sdk/worker"
25
"go.temporal.io/sdk/workflow"
26
"go.temporal.io/server/api/adminservice/v1"
27
"go.temporal.io/server/api/matchingservice/v1"
28
persistencespb "go.temporal.io/server/api/persistence/v1"
29
taskqueuespb "go.temporal.io/server/api/taskqueue/v1"
30
"go.temporal.io/server/chasm"
31
"go.temporal.io/server/common/dynamicconfig"
32
"go.temporal.io/server/common/searchattribute/sadefs"
33
"go.temporal.io/server/common/testing/parallelsuite"
34
"go.temporal.io/server/common/tqid"
35
"go.temporal.io/server/common/worker_versioning"
36
"go.temporal.io/server/tests/testcore"
37
"google.golang.org/protobuf/types/known/durationpb"
38
)
39
40
type VersioningIntegSuite struct {
41
parallelsuite.Suite[*VersioningIntegSuite]
42
}
43
44
const (
45
partitionTreeDegree = 3
46
longPollTime = 5 * time.Second
47
// use > 2 pollers by default to expose more timing situations
48
numPollers = 4
49
// These TTLs need to be greater than the time it takes for a workflow execution status change to show up in visibility
50
testReachabilityCacheOpenWFsTTL = 3 * time.Millisecond
51
testReachabilityCacheClosedWFsTTL = 6 * time.Millisecond
52
)
53
54
func TestVersioningFunctionalSuite(t *testing.T) {
55
parallelsuite.Run(t, &VersioningIntegSuite{})
56
}
57
58
func (s *VersioningIntegSuite) setupEnv(opts ...testcore.TestOption) *testcore.TestEnv {
59
opts = append([]testcore.TestOption{
60
testcore.WithDynamicConfig(dynamicconfig.EnableDeployments, true),
61
testcore.WithDynamicConfig(dynamicconfig.FrontendEnableWorkerVersioningDataAPIs, true),
62
testcore.WithDynamicConfig(dynamicconfig.FrontendEnableWorkerVersioningWorkflowAPIs, true),
63
testcore.WithDynamicConfig(dynamicconfig.FrontendEnableWorkerVersioningRuleAPIs, true),
64
testcore.WithDynamicConfig(dynamicconfig.TaskQueuesPerBuildIdLimit, 3),
65
66
testcore.WithDynamicConfig(dynamicconfig.AssignmentRuleLimitPerQueue, 10),
67
testcore.WithDynamicConfig(dynamicconfig.RedirectRuleLimitPerQueue, 10),
68
testcore.WithDynamicConfig(dynamicconfig.RedirectRuleMaxUpstreamBuildIDsPerQueue, 10),
69
testcore.WithDynamicConfig(dynamicconfig.MatchingDeletedRuleRetentionTime, 24*time.Hour),
70
testcore.WithDynamicConfig(dynamicconfig.ReachabilityBuildIdVisibilityGracePeriod, 3*time.Minute),
71
72
// Make sure we don't hit the rate limiter in tests
73
testcore.WithDynamicConfig(dynamicconfig.FrontendGlobalNamespaceNamespaceReplicationInducingAPIsRPS, 1000),
74
testcore.WithDynamicConfig(dynamicconfig.FrontendMaxNamespaceNamespaceReplicationInducingAPIsBurstRatioPerInstance, 1),
75
76
// The dispatch tests below rely on being able to see the effects of changing
77
// versioning data relatively quickly. In general, we only promise to act on new
78
// versioning data "soon", i.e. after a long poll interval. We can reduce the long poll
79
// interval so that we don't have to wait so long.
80
// TODO: after cleaning up old versioning we may not need to depend on long poll time. [cleanup-old-wv]
81
// This is because we don't interrupt long polls on change of versioning
82
// data, so if polls from different build ids go to the same matcher, old ones can steal
83
// tasks from new ones. If polls from different build ids never go to the same matcher
84
// anymore then we don't need it.
85
testcore.WithDynamicConfig(dynamicconfig.MatchingLongPollExpirationInterval, longPollTime),
86
87
// this is overridden for tests using testWithMatchingBehavior
88
testcore.WithDynamicConfig(dynamicconfig.MatchingNumTaskqueueReadPartitions, 4),
89
testcore.WithDynamicConfig(dynamicconfig.MatchingNumTaskqueueWritePartitions, 4),
90
91
// this is overridden since we don't want caching to be enabled while testing DescribeTaskQueue
92
// behaviour related to versioning
93
testcore.WithDynamicConfig(dynamicconfig.TaskQueueInfoByBuildIdTTL, 0*time.Second),
94
}, opts...)
95
96
return testcore.NewEnv(s.T(), opts...)
97
}
98
99
func (s *VersioningIntegSuite) reachabilityCacheTTLOptions() []testcore.TestOption {
100
return []testcore.TestOption{
101
testcore.WithDynamicConfig(dynamicconfig.ReachabilityCacheOpenWFsTTL, testReachabilityCacheOpenWFsTTL),
102
testcore.WithDynamicConfig(dynamicconfig.ReachabilityCacheClosedWFsTTL, testReachabilityCacheClosedWFsTTL),
103
}
104
}
105
106
func (s *VersioningIntegSuite) runTestWithMatchingBehavior(subtest func(*testcore.TestEnv, *VersioningIntegSuite)) {
107
for _, behavior := range testcore.AllMatchingBehaviors() {
108
s.Run(behavior.Name(), func(s *VersioningIntegSuite) {
109
env := s.setupEnv(behavior.Options()...)
110
behavior.InjectHooks(env)
111
subtest(env, s)
112
})
113
}
114
}
115
116
func (s *VersioningIntegSuite) TestVersionRuleConflictToken() {
117
env := s.setupEnv()
118
tq := "test-conflict-token"
119
120
// nil token --> fail
121
s.insertAssignmentRule(env, tq, "1", 0, nil, false)
122
123
// correct token from List --> success
124
cT1 := s.getVersioningRules(env, tq).GetConflictToken()
125
cT2 := s.insertAssignmentRule(env, tq, "2", 0, cT1, true)
126
127
// confirm token changed on insert but not on list
128
cT3 := s.getVersioningRules(env, tq).GetConflictToken()
129
s.NotEqual(cT1, cT2)
130
s.Equal(cT2, cT3)
131
132
// correct token from List or most recent mutation --> success
133
cT4 := s.insertAssignmentRule(env, tq, "3", 0, cT2, true)
134
s.getVersioningRules(env, tq)
135
136
// wrong token fails, same request with nil token also fails
137
s.insertAssignmentRule(env, tq, "4", 0, cT1, false)
138
s.insertAssignmentRule(env, tq, "4", 0, nil, false)
139
140
// wrong token fails, same request with correct token from Update succeeds
141
s.replaceAssignmentRule(env, tq, "20", 0, cT2, false)
142
cT5 := s.replaceAssignmentRule(env, tq, "20", 0, cT4, true)
143
cT6 := s.getVersioningRules(env, tq).GetConflictToken()
144
145
// confirm that list didn't change the conflict token again
146
s.Equal(cT5, cT6)
147
148
// wrong token fails, same request with correct token from List succeeds
149
s.deleteAssignmentRule(env, tq, 0, cT4, false)
150
s.deleteAssignmentRule(env, tq, 0, cT6, true)
151
152
// nil token fails
153
s.deleteAssignmentRule(env, tq, 0, nil, false)
154
}
155
156
func (s *VersioningIntegSuite) TestVersionRuleBuildIdValidation() {
157
// setup
158
env := s.setupEnv()
159
tq := "test-build-id-validation"
160
161
// get initial conflict token
162
cT := s.getVersioningRules(env, tq).GetConflictToken()
163
164
// failure due to long build id
165
longStr := strings.Repeat("0123456789", 30)
166
s.insertAssignmentRule(env, tq, longStr, 0, cT, false)
167
s.insertRedirectRule(env, tq, longStr, "foo", cT, false)
168
s.insertRedirectRule(env, tq, "foo", longStr, cT, false)
169
}
170
171
func (s *VersioningIntegSuite) TestAssignmentRuleInsert() {
172
// setup
173
env := s.setupEnv()
174
tq := "test-assignment-rule-insert"
175
176
// get initial conflict token
177
cT := s.getVersioningRules(env, tq).GetConflictToken()
178
179
// success
180
cT = s.insertAssignmentRule(env, tq, "1", 0, cT, true)
181
res1 := s.getVersioningRules(env, tq)
182
s.Equal("1", res1.GetAssignmentRules()[0].GetRule().GetTargetBuildId())
183
184
// failure due to out of bounds index
185
s.insertAssignmentRule(env, tq, "2", -1, cT, false)
186
s.Equal(res1, s.getVersioningRules(env, tq))
187
188
// success with conflict token returned by last successful call, same as above
189
s.insertAssignmentRule(env, tq, "2", 1, cT, true)
190
s.Equal("2", s.getVersioningRules(env, tq).GetAssignmentRules()[1].GetRule().GetTargetBuildId())
191
}
192
193
func (s *VersioningIntegSuite) TestAssignmentRuleReplace() {
194
// setup
195
env := s.setupEnv()
196
tq := "test-assignment-rule-replace"
197
198
// get initial conflict token + do initial inserts
199
cT := s.getVersioningRules(env, tq).GetConflictToken()
200
cT = s.insertAssignmentRule(env, tq, "1", 0, cT, true)
201
cT = s.insertAssignmentRule(env, tq, "2", 0, cT, true)
202
203
// success
204
cT = s.replaceAssignmentRule(env, tq, "3", 0, cT, true)
205
res := s.getVersioningRules(env, tq)
206
s.Equal("3", res.GetAssignmentRules()[0].GetRule().GetTargetBuildId())
207
208
// failure due to index out of bounds
209
s.replaceAssignmentRule(env, tq, "4", 10, cT, false)
210
s.Equal(res, s.getVersioningRules(env, tq))
211
212
// success with conflict token returned by last successful call, same as above
213
s.replaceAssignmentRule(env, tq, "4", 0, cT, true)
214
s.Equal("4", s.getVersioningRules(env, tq).GetAssignmentRules()[0].GetRule().GetTargetBuildId())
215
}
216
217
func (s *VersioningIntegSuite) TestAssignmentRuleDelete() {
218
// setup
219
env := s.setupEnv()
220
tq := "test-assignment-rule-delete"
221
222
// get initial conflict token + do initial inserts
223
cT := s.getVersioningRules(env, tq).GetConflictToken()
224
cT = s.insertAssignmentRule(env, tq, "1", 0, cT, true)
225
cT = s.insertAssignmentRule(env, tq, "2", 0, cT, true)
226
227
// success
228
cT = s.deleteAssignmentRule(env, tq, 0, cT, true)
229
res := s.getVersioningRules(env, tq)
230
s.Len(res.GetAssignmentRules(), 1)
231
232
// failure due to requirement that once a fully-ramped rule exists, at least one must always exist
233
s.deleteAssignmentRule(env, tq, 0, cT, false)
234
s.Equal(res, s.getVersioningRules(env, tq))
235
236
// insert another rule to prove that the conflict token was not the issue above
237
cT = s.insertAssignmentRule(env, tq, "2", 0, cT, true)
238
239
// delete again, success
240
s.deleteAssignmentRule(env, tq, 0, cT, true)
241
s.Len(res.GetAssignmentRules(), 1)
242
243
}
244
245
func (s *VersioningIntegSuite) TestRedirectRuleInsert() {
246
// setup
247
env := s.setupEnv()
248
tq := "test-redirect-rule-insert"
249
250
// get initial conflict token
251
cT := s.getVersioningRules(env, tq).GetConflictToken()
252
253
// success
254
cT = s.insertRedirectRule(env, tq, "1", "0", cT, true)
255
res := s.getVersioningRules(env, tq)
256
rulesMap := mkRedirectRulesMap(res.GetCompatibleRedirectRules())
257
s.Contains(rulesMap, "1")
258
s.Equal("0", rulesMap["1"])
259
260
// failure due to cycle
261
s.insertRedirectRule(env, tq, "0", "1", cT, false)
262
s.Equal(res, s.getVersioningRules(env, tq))
263
264
// success with same conflict token but no cycle
265
s.insertRedirectRule(env, tq, "0", "2", cT, true)
266
}
267
268
func (s *VersioningIntegSuite) TestRedirectRuleReplace() {
269
// setup
270
env := s.setupEnv()
271
tq := "test-redirect-rule-replace"
272
273
// get initial conflict token + do initial insert
274
cT := s.getVersioningRules(env, tq).GetConflictToken()
275
cT = s.insertRedirectRule(env, tq, "1", "0", cT, true)
276
277
// success
278
cT = s.replaceRedirectRule(env, tq, "1", "2", cT, true)
279
res := s.getVersioningRules(env, tq)
280
rulesMap := mkRedirectRulesMap(res.GetCompatibleRedirectRules())
281
s.Contains(rulesMap, "1")
282
s.Equal("2", rulesMap["1"])
283
284
// failure due to source not found
285
s.replaceRedirectRule(env, tq, "10", "3", cT, false)
286
s.Equal(res, s.getVersioningRules(env, tq))
287
288
// success with same conflict token and correct source
289
s.replaceRedirectRule(env, tq, "1", "3", cT, true)
290
}
291
292
func (s *VersioningIntegSuite) TestRedirectRuleDelete() {
293
// setup
294
env := s.setupEnv()
295
tq := "test-redirect-rule-delete"
296
297
// get initial conflict token + do initial inserts
298
cT := s.getVersioningRules(env, tq).GetConflictToken()
299
cT = s.insertRedirectRule(env, tq, "1", "0", cT, true)
300
cT = s.insertRedirectRule(env, tq, "2", "0", cT, true)
301
302
// success
303
cT = s.deleteRedirectRule(env, tq, "1", cT, true)
304
res := s.getVersioningRules(env, tq)
305
s.Len(res.GetCompatibleRedirectRules(), 1)
306
307
// failure due to source not found
308
s.deleteRedirectRule(env, tq, "1", cT, false)
309
s.Equal(res, s.getVersioningRules(env, tq))
310
311
// success with same conflict token and valid source
312
s.deleteRedirectRule(env, tq, "2", cT, true)
313
}
314
315
func (s *VersioningIntegSuite) TestCommitBuildID() {
316
// setup
317
env := s.setupEnv()
318
tq := "test-commit-build-id"
319
320
// get initial conflict token
321
cT := s.getVersioningRules(env, tq).GetConflictToken()
322
323
// no recent poller --> failure
324
s.commitBuildID(env, tq, "1", false, cT, false)
325
326
// no recent poller + force --> success
327
cT = s.commitBuildID(env, tq, "1", true, cT, true)
328
res := s.getVersioningRules(env, tq)
329
s.Len(res.GetAssignmentRules(), 1)
330
s.Empty(res.GetCompatibleRedirectRules())
331
s.Equal("1", res.GetAssignmentRules()[0].GetRule().GetTargetBuildId())
332
s.Equal(float32(100), res.GetAssignmentRules()[0].GetRule().GetPercentageRamp().GetRampPercentage())
333
334
// recent versioned poller on wrong build ID --> failure
335
s.registerWorkflowAndPollVersionedTaskQueue(env, tq, "3", true)
336
s.commitBuildID(env, tq, "2", false, cT, false)
337
338
// recent unversioned poller on build ID 2 --> failure
339
s.registerWorkflowAndPollVersionedTaskQueue(env, tq, "2", false)
340
s.commitBuildID(env, tq, "2", false, cT, false)
341
342
// recent versioned poller on build ID 2 --> success
343
s.registerWorkflowAndPollVersionedTaskQueue(env, tq, "2", true)
344
s.commitBuildID(env, tq, "2", false, cT, true)
345
res = s.getVersioningRules(env, tq)
346
s.Len(res.GetAssignmentRules(), 1)
347
s.Empty(res.GetCompatibleRedirectRules())
348
s.Equal("2", res.GetAssignmentRules()[0].GetRule().GetTargetBuildId())
349
s.Equal(float32(100), res.GetAssignmentRules()[0].GetRule().GetPercentageRamp().GetRampPercentage())
350
}
351
352
func mkRedirectRulesMap(redirectRules []*taskqueuepb.TimestampedCompatibleBuildIdRedirectRule) map[string]string {
353
ret := make(map[string]string)
354
for _, r := range redirectRules {
355
rule := r.GetRule()
356
ret[rule.GetSourceBuildId()] = rule.GetTargetBuildId()
357
}
358
return ret
359
}
360
361
func (s *VersioningIntegSuite) TestBasicVersionUpdate() {
362
env := s.setupEnv()
363
tq := "functional-versioning-basic"
364
365
foo := s.prefixed("foo")
366
s.addNewDefaultBuildID(env, tq, foo)
367
368
res2, err := env.FrontendClient().GetWorkerBuildIdCompatibility(s.Context(), &workflowservice.GetWorkerBuildIdCompatibilityRequest{
369
Namespace: env.Namespace().String(),
370
TaskQueue: tq,
371
})
372
s.NoError(err)
373
s.NotNil(res2)
374
s.Equal(foo, getCurrentDefault(res2))
375
}
376
377
func (s *VersioningIntegSuite) TestSeriesOfUpdates() {
378
env := s.setupEnv()
379
tq := "functional-versioning-series"
380
381
for i := range 10 {
382
s.addNewDefaultBuildID(env, tq, s.prefixed(fmt.Sprintf("foo-%d", i)))
383
}
384
s.addCompatibleBuildID(env, tq, s.prefixed("foo-2.1"), s.prefixed("foo-2"), false)
385
386
res, err := env.FrontendClient().GetWorkerBuildIdCompatibility(s.Context(), &workflowservice.GetWorkerBuildIdCompatibilityRequest{
387
Namespace: env.Namespace().String(),
388
TaskQueue: tq,
389
})
390
s.NoError(err)
391
s.NotNil(res)
392
s.Equal(s.prefixed("foo-9"), getCurrentDefault(res))
393
s.Equal(s.prefixed("foo-2.1"), res.GetMajorVersionSets()[2].GetBuildIds()[1])
394
s.Equal(s.prefixed("foo-2"), res.GetMajorVersionSets()[2].GetBuildIds()[0])
395
}
396
397
func (s *VersioningIntegSuite) TestLinkToNonexistentCompatibleVersionReturnsNotFound() {
398
env := s.setupEnv()
399
tq := "functional-versioning-compat-not-found"
400
401
res, err := env.FrontendClient().UpdateWorkerBuildIdCompatibility(s.Context(), &workflowservice.UpdateWorkerBuildIdCompatibilityRequest{
402
Namespace: env.Namespace().String(),
403
TaskQueue: tq,
404
Operation: &workflowservice.UpdateWorkerBuildIdCompatibilityRequest_AddNewCompatibleBuildId{
405
AddNewCompatibleBuildId: &workflowservice.UpdateWorkerBuildIdCompatibilityRequest_AddNewCompatibleVersion{
406
NewBuildId: "foo",
407
ExistingCompatibleBuildId: "i don't exist yo",
408
},
409
},
410
})
411
s.Error(err)
412
s.Nil(res)
413
s.IsType(&serviceerror.NotFound{}, err)
414
}
415
416
func (s *VersioningIntegSuite) TestVersioningStatePersistsAcrossUnload() {
417
env := s.setupEnv()
418
tq := "functional-versioning-persists"
419
420
s.addNewDefaultBuildID(env, tq, s.prefixed("foo"))
421
422
// Unload task queue to make sure the data is there when we load it again.
423
s.unloadTaskQueue(env, tq)
424
425
res, err := env.FrontendClient().GetWorkerBuildIdCompatibility(s.Context(), &workflowservice.GetWorkerBuildIdCompatibilityRequest{
426
Namespace: env.Namespace().String(),
427
TaskQueue: tq,
428
})
429
s.NoError(err)
430
s.NotNil(res)
431
s.Equal(s.prefixed("foo"), getCurrentDefault(res))
432
}
433
434
func (s *VersioningIntegSuite) TestVersioningChangesPropagate() {
435
tq := "functional-versioning-propagate"
436
437
// ensure at least two hops
438
const partCount = 1 + partitionTreeDegree + partitionTreeDegree*partitionTreeDegree
439
env := s.setupEnv(
440
testcore.WithDynamicConfig(dynamicconfig.MatchingNumTaskqueueReadPartitions, partCount),
441
testcore.WithDynamicConfig(dynamicconfig.MatchingNumTaskqueueWritePartitions, partCount),
442
)
443
444
for _, buildID := range []string{"foo", "foo-v2", "foo-v3"} {
445
s.addNewDefaultBuildID(env, tq, buildID)
446
s.waitForVersionSetPropagation(env, tq, buildID)
447
}
448
}
449
450
func (s *VersioningIntegSuite) TestMaxTaskQueuesPerBuildIdEnforced() {
451
env := s.setupEnv()
452
buildID := fmt.Sprintf("b-%s", s.T().Name())
453
// Map a 3 task queues to this build ID and verify success
454
for i := 1; i <= 3; i++ {
455
taskQueue := fmt.Sprintf("q-%s-%d", s.T().Name(), i)
456
_, err := env.FrontendClient().UpdateWorkerBuildIdCompatibility(s.Context(), &workflowservice.UpdateWorkerBuildIdCompatibilityRequest{
457
Namespace: env.Namespace().String(),
458
TaskQueue: taskQueue,
459
Operation: &workflowservice.UpdateWorkerBuildIdCompatibilityRequest_AddNewBuildIdInNewDefaultSet{
460
AddNewBuildIdInNewDefaultSet: buildID,
461
},
462
})
463
s.NoError(err)
464
}
465
466
// Map a fourth task queue to this build ID and verify it errors
467
taskQueue := fmt.Sprintf("q-%s-%d", s.T().Name(), 4)
468
_, err := env.FrontendClient().UpdateWorkerBuildIdCompatibility(s.Context(), &workflowservice.UpdateWorkerBuildIdCompatibilityRequest{
469
Namespace: env.Namespace().String(),
470
TaskQueue: taskQueue,
471
Operation: &workflowservice.UpdateWorkerBuildIdCompatibilityRequest_AddNewBuildIdInNewDefaultSet{
472
AddNewBuildIdInNewDefaultSet: buildID,
473
},
474
})
475
var failedPreconditionError *serviceerror.FailedPrecondition
476
s.ErrorAs(err, &failedPreconditionError)
477
s.Equal("Exceeded max task queues allowed to be mapped to a single build ID: 3", failedPreconditionError.Message)
478
}
479
480
func (s *VersioningIntegSuite) TestDispatchNewWorkflowOld() {
481
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) { s.dispatchNewWorkflow(env, false) })
482
}
483
484
func (s *VersioningIntegSuite) TestDispatchNewWorkflow() {
485
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) { s.dispatchNewWorkflow(env, true) })
486
}
487
488
func (s *VersioningIntegSuite) dispatchNewWorkflow(
489
env *testcore.TestEnv,
490
newVersioning bool,
491
) {
492
tq := testcore.RandomizeStr(s.T().Name())
493
v1 := s.prefixed("v1")
494
495
wf := func(ctx workflow.Context) (string, error) {
496
return "done!", nil
497
}
498
499
if newVersioning {
500
rule := s.addAssignmentRule(env, tq, v1)
501
s.waitForAssignmentRulePropagation(env, tq, rule)
502
} else {
503
s.addNewDefaultBuildID(env, tq, v1)
504
s.waitForVersionSetPropagation(env, tq, v1)
505
}
506
507
w1 := worker.New(env.SdkClient(), tq, worker.Options{
508
BuildID: v1,
509
UseBuildIDForVersioning: true,
510
MaxConcurrentWorkflowTaskPollers: numPollers,
511
})
512
w1.RegisterWorkflow(wf)
513
s.NoError(w1.Start())
514
defer w1.Stop()
515
516
run, err := env.SdkClient().ExecuteWorkflow(s.Context(), sdkclient.StartWorkflowOptions{TaskQueue: tq}, wf)
517
s.NoError(err)
518
var out string
519
s.NoError(run.Get(s.Context(), &out))
520
s.Equal("done!", out)
521
522
s.validateWorkflowBuildIds(env, run.GetID(), run.GetRunID(), v1, newVersioning, v1, "", nil)
523
if newVersioning {
524
s.validateWorkflowEventsVersionStamps(env, run.GetID(), run.GetRunID(), []string{v1}, "")
525
} else {
526
s.validateWorkflowEventsVersionStamps(env, run.GetID(), run.GetRunID(), []string{}, "")
527
}
528
}
529
530
func (s *VersioningIntegSuite) TestDispatchNewWorkflowWithRamp() {
531
env := s.setupEnv()
532
tq := testcore.RandomizeStr(s.T().Name())
533
v1 := s.prefixed("v1")
534
v2 := s.prefixed("v2")
535
536
wf1 := func(ctx workflow.Context) (string, error) {
537
return "done v1!", nil
538
}
539
wf2 := func(ctx workflow.Context) (string, error) {
540
return "done v2!", nil
541
}
542
543
rule := s.addAssignmentRule(env, tq, v1)
544
rule2 := s.addAssignmentRuleWithRamp(env, tq, v2, 50)
545
s.waitForAssignmentRulePropagation(env, tq, rule)
546
s.waitForAssignmentRulePropagation(env, tq, rule2)
547
548
w1 := worker.New(env.SdkClient(), tq, worker.Options{
549
BuildID: v1,
550
UseBuildIDForVersioning: true,
551
MaxConcurrentWorkflowTaskPollers: numPollers,
552
})
553
w1.RegisterWorkflowWithOptions(wf1, workflow.RegisterOptions{Name: "wf"})
554
s.NoError(w1.Start())
555
defer w1.Stop()
556
w2 := worker.New(env.SdkClient(), tq, worker.Options{
557
BuildID: v2,
558
UseBuildIDForVersioning: true,
559
MaxConcurrentWorkflowTaskPollers: numPollers,
560
})
561
w2.RegisterWorkflowWithOptions(wf2, workflow.RegisterOptions{Name: "wf"})
562
s.NoError(w2.Start())
563
defer w2.Stop()
564
565
counter := make(map[string]int)
566
for range 50 {
567
run, err := env.SdkClient().ExecuteWorkflow(s.Context(), sdkclient.StartWorkflowOptions{TaskQueue: tq}, "wf")
568
s.NoError(err)
569
var out string
570
s.NoError(run.Get(s.Context(), &out))
571
counter[out]++
572
}
573
574
// both builds should've got executions
575
s.Positive(counter["done v1!"])
576
s.Positive(counter["done v2!"])
577
s.Equal(50, counter["done v1!"]+counter["done v2!"])
578
}
579
580
func (s *VersioningIntegSuite) TestWorkflowStaysInBuildId() {
581
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) { s.workflowStaysInBuildID(env) })
582
}
583
584
func (s *VersioningIntegSuite) workflowStaysInBuildID(env *testcore.TestEnv) {
585
tq := testcore.RandomizeStr(s.T().Name())
586
v1 := s.prefixed("v1")
587
v2 := s.prefixed("v2")
588
589
act1Done := make(chan struct{})
590
rulesUpdated := make(chan struct{})
591
592
act1 := func() (string, error) {
593
close(act1Done)
594
return "act1 done!", nil
595
}
596
597
act2 := func() (string, error) {
598
env.WaitForChannel(rulesUpdated)
599
return "act2 done!", nil
600
}
601
602
wf := func(ctx workflow.Context) (string, error) {
603
var ret string
604
err := workflow.ExecuteActivity(workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
605
DisableEagerExecution: true,
606
StartToCloseTimeout: 1 * time.Second,
607
}), act1).Get(ctx, &ret)
608
s.NoError(err)
609
s.Equal("act1 done!", ret)
610
611
// assignment rules change in here
612
613
err = workflow.ExecuteActivity(workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
614
DisableEagerExecution: true,
615
StartToCloseTimeout: 1 * time.Second,
616
}), act2).Get(ctx, &ret)
617
s.NoError(err)
618
s.Equal("act2 done!", ret)
619
return "done!", nil
620
}
621
622
rule := s.addAssignmentRule(env, tq, v1)
623
s.waitForAssignmentRulePropagation(env, tq, rule)
624
625
w1 := worker.New(env.SdkClient(), tq, worker.Options{
626
BuildID: v1,
627
UseBuildIDForVersioning: true,
628
MaxConcurrentWorkflowTaskPollers: numPollers,
629
})
630
w1.RegisterWorkflow(wf)
631
w1.RegisterActivity(act1)
632
w1.RegisterActivity(act2)
633
s.NoError(w1.Start())
634
defer w1.Stop()
635
636
run, err := env.SdkClient().ExecuteWorkflow(s.Context(), sdkclient.StartWorkflowOptions{TaskQueue: tq}, wf)
637
s.NoError(err)
638
639
env.WaitForChannel(act1Done)
640
s.validateWorkflowBuildIds(env, run.GetID(), run.GetRunID(), v1, true, v1, "", nil)
641
642
// update rules with v2 as the default build
643
rule = s.addAssignmentRule(env, tq, v2)
644
s.waitForAssignmentRulePropagation(env, tq, rule)
645
646
dw, err := env.SdkClient().DescribeWorkflowExecution(s.Context(), run.GetID(), run.GetRunID())
647
s.NoError(err)
648
s.Len(dw.GetPendingActivities(), 1)
649
s.NotNil(dw.GetPendingActivities()[0].GetUseWorkflowBuildId())
650
651
close(rulesUpdated)
652
var out string
653
s.NoError(run.Get(s.Context(), &out))
654
s.Equal("done!", out)
655
s.validateWorkflowBuildIds(env, run.GetID(), run.GetRunID(), v1, true, v1, "", nil)
656
s.validateWorkflowEventsVersionStamps(env, run.GetID(), run.GetRunID(), []string{
657
v1, // WFT
658
v1, // activity
659
// v1, skipped because it belongs to sticky queue
660
v1, // activity
661
// v1, skipped because it belongs to sticky queue
662
}, "")
663
}
664
665
func (s *VersioningIntegSuite) TestUnversionedWorkflowStaysUnversioned() {
666
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) { s.unversionedWorkflowStaysUnversioned(env) })
667
}
668
669
func (s *VersioningIntegSuite) unversionedWorkflowStaysUnversioned(env *testcore.TestEnv) {
670
tq := testcore.RandomizeStr(s.T().Name())
671
v1 := s.prefixed("v1")
672
673
act1Done := make(chan struct{})
674
rulesUpdated := make(chan struct{})
675
676
act1 := func() (string, error) {
677
close(act1Done)
678
return "act1 done!", nil
679
}
680
681
act2 := func() (string, error) {
682
env.WaitForChannel(rulesUpdated)
683
return "act2 done!", nil
684
}
685
686
wf := func(ctx workflow.Context) (string, error) {
687
var ret string
688
err := workflow.ExecuteActivity(workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
689
DisableEagerExecution: true,
690
StartToCloseTimeout: 1 * time.Second,
691
}), act1).Get(ctx, &ret)
692
s.NoError(err)
693
s.Equal("act1 done!", ret)
694
695
// assignment rules change in here
696
697
err = workflow.ExecuteActivity(workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
698
DisableEagerExecution: true,
699
StartToCloseTimeout: 1 * time.Second,
700
}), act2).Get(ctx, &ret)
701
s.NoError(err)
702
s.Equal("act2 done!", ret)
703
return "done!", nil
704
}
705
706
w1 := worker.New(env.SdkClient(), tq, worker.Options{
707
MaxConcurrentWorkflowTaskPollers: numPollers,
708
})
709
w1.RegisterWorkflow(wf)
710
w1.RegisterActivity(act1)
711
w1.RegisterActivity(act2)
712
s.NoError(w1.Start())
713
defer w1.Stop()
714
715
run, err := env.SdkClient().ExecuteWorkflow(s.Context(), sdkclient.StartWorkflowOptions{TaskQueue: tq}, wf)
716
s.NoError(err)
717
718
env.WaitForChannel(act1Done)
719
s.validateWorkflowBuildIds(env, run.GetID(), run.GetRunID(), "", true, "binary-checksum", "", nil)
720
721
// update rules with v1 as the default build
722
rule := s.addAssignmentRule(env, tq, v1)
723
s.waitForAssignmentRulePropagation(env, tq, rule)
724
725
dw, err := env.SdkClient().DescribeWorkflowExecution(s.Context(), run.GetID(), run.GetRunID())
726
s.NoError(err)
727
s.Len(dw.GetPendingActivities(), 1)
728
s.Nil(dw.GetPendingActivities()[0].GetAssignedBuildId())
729
close(rulesUpdated)
730
731
var out string
732
s.NoError(run.Get(s.Context(), &out))
733
s.Equal("done!", out)
734
s.validateWorkflowBuildIds(env, run.GetID(), run.GetRunID(), "", true, "binary-checksum", "", nil)
735
s.validateWorkflowEventsVersionStamps(env, run.GetID(), run.GetRunID(), []string{"", "", "", "", ""}, "")
736
}
737
738
func (s *VersioningIntegSuite) TestFirstWorkflowTaskAssignment_Spooled() {
739
s.T().Skip("Skipping test since rules based versioning is soon to be deprecated")
740
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) { s.firstWorkflowTaskAssignmentSpooled(env) })
741
}
742
743
func (s *VersioningIntegSuite) firstWorkflowTaskAssignmentSpooled(env *testcore.TestEnv) {
744
tq := testcore.RandomizeStr(s.T().Name())
745
v1 := s.prefixed("v1")
746
v2 := s.prefixed("v2")
747
v3 := s.prefixed("v3")
748
749
rule := s.addAssignmentRule(env, tq, v1)
750
s.waitForAssignmentRulePropagation(env, tq, rule)
751
752
// start wf without worker
753
run, err := env.SdkClient().ExecuteWorkflow(s.Context(), sdkclient.StartWorkflowOptions{
754
TaskQueue: tq,
755
// setting to a short timeout to verify first wf task timeout assignment
756
WorkflowTaskTimeout: 1 * time.Second,
757
}, "wf")
758
s.NoError(err)
759
760
// MS should have the correct build ID
761
s.waitForWorkflowBuildID(env, run.GetID(), run.GetRunID(), v1)
762
s.validateWorkflowBuildIds(env, run.GetID(), run.GetRunID(), v1, true, "", "", nil)
763
764
// update latest build to v2
765
rule = s.addAssignmentRule(env, tq, v2)
766
s.waitForAssignmentRulePropagation(env, tq, rule)
767
768
failedTask := make(chan struct{})
769
wf1 := func(ctx workflow.Context) (string, error) {
770
close(failedTask)
771
panic("failing WF task intentionally")
772
}
773
774
// run worker on v1 so it can fail the scheduled task
775
w1 := worker.New(env.SdkClient(), tq, worker.Options{
776
BuildID: v1,
777
UseBuildIDForVersioning: true,
778
MaxConcurrentWorkflowTaskPollers: numPollers,
779
})
780
w1.RegisterWorkflowWithOptions(wf1, workflow.RegisterOptions{Name: "wf"})
781
s.NoError(w1.Start())
782
defer w1.Stop()
783
784
env.WaitForChannel(failedTask)
785
786
// After scheduling the second time, now MS should be assigned to v2
787
s.waitForWorkflowBuildID(env, run.GetID(), run.GetRunID(), v2)
788
s.validateWorkflowBuildIds(env, run.GetID(), run.GetRunID(), v2, true, "", "", []string{v1})
789
790
// update latest build to v3
791
rule = s.addAssignmentRule(env, tq, v3)
792
s.waitForAssignmentRulePropagation(env, tq, rule)
793
794
timedoutTask := make(chan struct{})
795
wf2 := func(ctx workflow.Context) (string, error) {
796
time.Sleep(1 * time.Second) //nolint:forbidigo
797
timedoutTask <- struct{}{}
798
time.Sleep(100 * time.Second) //nolint:forbidigo
799
return "return after long sleep", nil
800
}
801
802
// run worker on v2 so it can timeout the scheduled task
803
w2 := worker.New(env.SdkClient(), tq, worker.Options{
804
BuildID: v2,
805
UseBuildIDForVersioning: true,
806
MaxConcurrentWorkflowTaskPollers: numPollers,
807
// since the WF cache is shared by all workers in this process, we need to set this in a way that ensures
808
// WFTs will timeout in w2 but not in w3
809
DeadlockDetectionTimeout: 1500 * time.Millisecond,
810
})
811
w2.RegisterWorkflowWithOptions(wf2, workflow.RegisterOptions{Name: "wf"})
812
s.NoError(w2.Start())
813
defer w2.Stop()
814
815
env.WaitForChannel(timedoutTask)
816
817
// After scheduling the third time, now MS should be assigned to v3
818
s.waitForWorkflowBuildID(env, run.GetID(), run.GetRunID(), v3)
819
s.validateWorkflowBuildIds(env, run.GetID(), run.GetRunID(), v3, true, "", "", []string{v1, v2})
820
821
wf3 := func(ctx workflow.Context) (string, error) {
822
return "done on v3!", nil
823
}
824
825
// run worker on v3 so it can complete the wf
826
w3 := worker.New(env.SdkClient(), tq, worker.Options{
827
BuildID: v3,
828
UseBuildIDForVersioning: true,
829
MaxConcurrentWorkflowTaskPollers: numPollers,
830
})
831
w3.RegisterWorkflowWithOptions(wf3, workflow.RegisterOptions{Name: "wf"})
832
s.NoError(w3.Start())
833
defer w3.Stop()
834
835
var out string
836
s.NoError(run.Get(s.Context(), &out))
837
s.Equal("done on v3!", out)
838
s.validateWorkflowBuildIds(env, run.GetID(), run.GetRunID(), v3, true, v3, "", []string{v1, v2})
839
s.validateWorkflowEventsVersionStamps(env,
840
run.GetID(), run.GetRunID(), []string{
841
v1, // failed wf tasks
842
v2, // timed out wf tasks
843
v3, // succeeded wf task
844
}, "",
845
)
846
}
847
848
func (s *VersioningIntegSuite) TestFirstWorkflowTaskAssignment_SyncMatch() {
849
s.T().Skip("Skipping test since rules based versioning is soon to be deprecated")
850
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) { s.firstWorkflowTaskAssignmentSyncMatch(env) })
851
}
852
853
func (s *VersioningIntegSuite) firstWorkflowTaskAssignmentSyncMatch(env *testcore.TestEnv) {
854
tq := testcore.RandomizeStr(s.T().Name())
855
v1 := s.prefixed("v1")
856
v2 := s.prefixed("v2")
857
v3 := s.prefixed("v3")
858
859
rule := s.addAssignmentRule(env, tq, v1)
860
s.waitForAssignmentRulePropagation(env, tq, rule)
861
862
// v1 fails the task
863
failedTask := make(chan struct{})
864
wf1 := func(ctx workflow.Context) (string, error) {
865
failedTask <- struct{}{}
866
panic("failing WF task intentionally")
867
}
868
869
w1 := worker.New(env.SdkClient(), tq, worker.Options{
870
BuildID: v1,
871
UseBuildIDForVersioning: true,
872
MaxConcurrentWorkflowTaskPollers: numPollers,
873
})
874
w1.RegisterWorkflowWithOptions(wf1, workflow.RegisterOptions{Name: "wf"})
875
s.NoError(w1.Start())
876
defer w1.Stop()
877
878
run, err := env.SdkClient().ExecuteWorkflow(s.Context(), sdkclient.StartWorkflowOptions{
879
TaskQueue: tq,
880
// setting to a short timeout to verify first wf task timeout assignment
881
WorkflowTaskTimeout: 1 * time.Second,
882
}, "wf")
883
s.NoError(err)
884
885
// wait for two failures to make sure more attempts does not generate more history tasks
886
env.WaitForChannel(failedTask)
887
env.WaitForChannel(failedTask)
888
889
// MS should have the correct build ID
890
s.validateWorkflowBuildIds(env, run.GetID(), run.GetRunID(), v1, true, "", "", nil)
891
892
// v2 times out the task
893
timedoutTask := make(chan struct{})
894
wf2 := func(ctx workflow.Context) (string, error) {
895
time.Sleep(1 * time.Second) //nolint:forbidigo
896
timedoutTask <- struct{}{}
897
time.Sleep(100 * time.Second) //nolint:forbidigo
898
return "return after long sleep", nil
899
}
900
901
w2 := worker.New(env.SdkClient(), tq, worker.Options{
902
BuildID: v2,
903
UseBuildIDForVersioning: true,
904
MaxConcurrentWorkflowTaskPollers: numPollers,
905
// since the WF cache is shared by all workers in this process, we need to set this in a way that ensures
906
// WFTs will timeout in w2 but not in w3
907
DeadlockDetectionTimeout: 1500 * time.Millisecond,
908
})
909
w2.RegisterWorkflowWithOptions(wf2, workflow.RegisterOptions{Name: "wf"})
910
s.NoError(w2.Start())
911
defer w2.Stop()
912
913
// update latest build to v2
914
rule = s.addAssignmentRule(env, tq, v2)
915
s.waitForAssignmentRulePropagation(env, tq, rule)
916
917
// wait for multiple timeouts to make sure more attempts do not generate more history events
918
env.WaitForChannel(timedoutTask)
919
env.WaitForChannel(timedoutTask)
920
env.WaitForChannel(timedoutTask)
921
922
// After scheduling the second time, now MS should be assigned to v2
923
s.validateWorkflowBuildIds(env, run.GetID(), run.GetRunID(), v2, true, "", "", []string{v1})
924
925
// v3 can process the task
926
wf3 := func(ctx workflow.Context) (string, error) {
927
return "done on v3!", nil
928
}
929
930
// run worker on v3 so it can complete the wf
931
w3 := worker.New(env.SdkClient(), tq, worker.Options{
932
BuildID: v3,
933
UseBuildIDForVersioning: true,
934
MaxConcurrentWorkflowTaskPollers: numPollers,
935
})
936
w3.RegisterWorkflowWithOptions(wf3, workflow.RegisterOptions{Name: "wf"})
937
s.NoError(w3.Start())
938
defer w3.Stop()
939
940
// update latest build to v3
941
rule = s.addAssignmentRule(env, tq, v3)
942
s.waitForAssignmentRulePropagation(env, tq, rule)
943
944
var out string
945
s.NoError(run.Get(s.Context(), &out))
946
s.Equal("done on v3!", out)
947
s.validateWorkflowBuildIds(env, run.GetID(), run.GetRunID(), v3, true, v3, "", []string{v1, v2})
948
s.validateWorkflowEventsVersionStamps(env,
949
run.GetID(), run.GetRunID(), []string{
950
v1, // failed wf tasks
951
v2, // timed out wf tasks
952
v3, // succeeded wf task
953
}, "",
954
)
955
}
956
957
func (s *VersioningIntegSuite) TestIndependentActivityTaskAssignment_Spooled_VersionedWorkflow() {
958
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) {
959
s.independentActivityTaskAssignmentSpooled(env, true)
960
})
961
}
962
963
func (s *VersioningIntegSuite) TestIndependentActivityTaskAssignment_Spooled_UnversionedWorkflow() {
964
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) {
965
s.independentActivityTaskAssignmentSpooled(env, false)
966
})
967
}
968
969
func (s *VersioningIntegSuite) independentActivityTaskAssignmentSpooled(
970
env *testcore.TestEnv,
971
versionedWf bool,
972
) {
973
wfTq := testcore.RandomizeStr(s.T().Name())
974
actTq := testcore.RandomizeStr(s.T().Name())
975
wfV1 := s.prefixed("wfv1")
976
v1 := s.prefixed("v1")
977
v2 := s.prefixed("v2")
978
v3 := s.prefixed("v3")
979
980
rule := s.addAssignmentRule(env, actTq, v1)
981
s.waitForAssignmentRulePropagation(env, actTq, rule)
982
if versionedWf {
983
rule = s.addAssignmentRule(env, wfTq, wfV1)
984
s.waitForAssignmentRulePropagation(env, wfTq, rule)
985
}
986
987
wf := func(ctx workflow.Context) (string, error) {
988
var ret string
989
err := workflow.ExecuteActivity(workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
990
TaskQueue: actTq,
991
StartToCloseTimeout: 1 * time.Second,
992
RetryPolicy: &temporal.RetryPolicy{
993
InitialInterval: 1 * time.Second,
994
BackoffCoefficient: 1,
995
},
996
}), "act").Get(ctx, &ret)
997
s.NoError(err)
998
s.Equal("done in v3!", ret)
999
return "done!", nil
1000
}
1001
1002
// start workflow worker
1003
wfw := worker.New(env.SdkClient(), wfTq, worker.Options{
1004
BuildID: wfV1,
1005
UseBuildIDForVersioning: versionedWf,
1006
MaxConcurrentWorkflowTaskPollers: numPollers,
1007
})
1008
wfw.RegisterWorkflow(wf)
1009
s.NoError(wfw.Start())
1010
defer wfw.Stop()
1011
1012
// start wf without activity worker
1013
run, err := env.SdkClient().ExecuteWorkflow(s.Context(), sdkclient.StartWorkflowOptions{
1014
TaskQueue: wfTq,
1015
// setting to a short timeout to verify first wf task timeout assignment
1016
WorkflowTaskTimeout: 1 * time.Second,
1017
}, wf)
1018
s.NoError(err)
1019
1020
// MS should have the correct build ID after finishing the first WFT
1021
s.Eventually(
1022
func() bool {
1023
dw, err := env.SdkClient().DescribeWorkflowExecution(s.Context(), run.GetID(), run.GetRunID())
1024
s.NoError(err)
1025
if len(dw.GetPendingActivities()) == 0 {
1026
return false
1027
}
1028
if versionedWf {
1029
s.Equal(wfV1, dw.GetWorkflowExecutionInfo().GetAssignedBuildId())
1030
s.Equal(wfV1, dw.GetWorkflowExecutionInfo().GetMostRecentWorkerVersionStamp().GetBuildId())
1031
} else {
1032
s.Empty(dw.GetWorkflowExecutionInfo().GetAssignedBuildId()) //nolint:staticcheck
1033
s.False(dw.GetWorkflowExecutionInfo().GetMostRecentWorkerVersionStamp().GetUseVersioning())
1034
}
1035
return v1 == dw.GetPendingActivities()[0].GetLastIndependentlyAssignedBuildId()
1036
},
1037
10*time.Second,
1038
50*time.Millisecond,
1039
)
1040
1041
// update latest build to v2
1042
rule = s.addAssignmentRule(env, actTq, v2)
1043
s.waitForAssignmentRulePropagation(env, actTq, rule)
1044
1045
failedTask := make(chan struct{})
1046
act1 := func() (string, error) {
1047
close(failedTask)
1048
return "", errors.New("failing activity task intentionally") //nolint:err113
1049
}
1050
1051
// run v1 activity worker so it can fail the scheduled activity
1052
w1 := worker.New(env.SdkClient(), actTq, worker.Options{
1053
BuildID: v1,
1054
UseBuildIDForVersioning: true,
1055
MaxConcurrentWorkflowTaskPollers: numPollers,
1056
})
1057
w1.RegisterActivityWithOptions(act1, activity.RegisterOptions{Name: "act"})
1058
s.NoError(w1.Start())
1059
defer w1.Stop()
1060
1061
env.WaitForChannel(failedTask)
1062
1063
// After scheduling the second time, now pending activity should be assigned to v2
1064
s.Eventually(
1065
func() bool {
1066
dw, err := env.SdkClient().DescribeWorkflowExecution(s.Context(), run.GetID(), run.GetRunID())
1067
s.NoError(err)
1068
s.Len(dw.GetPendingActivities(), 1)
1069
return v2 == dw.GetPendingActivities()[0].GetLastIndependentlyAssignedBuildId()
1070
},
1071
10*time.Second,
1072
50*time.Millisecond,
1073
)
1074
1075
// update latest build to v3
1076
rule = s.addAssignmentRule(env, actTq, v3)
1077
s.waitForAssignmentRulePropagation(env, actTq, rule)
1078
1079
timedoutTask := make(chan struct{})
1080
act2 := func() (string, error) {
1081
time.Sleep(1 * time.Second) //nolint:forbidigo
1082
close(timedoutTask)
1083
time.Sleep(100 * time.Second) //nolint:forbidigo
1084
return "return after long sleep", nil
1085
}
1086
1087
// run v2 activity worker so it can timeout the scheduled activity
1088
w2 := worker.New(env.SdkClient(), actTq, worker.Options{
1089
BuildID: v2,
1090
UseBuildIDForVersioning: true,
1091
MaxConcurrentWorkflowTaskPollers: numPollers,
1092
// since the WF cache is shared by all workers in this process, we need to set this in a way that ensures
1093
// WFTs will timeout in w2 but not in w3
1094
DeadlockDetectionTimeout: 1500 * time.Millisecond,
1095
})
1096
w2.RegisterActivityWithOptions(act2, activity.RegisterOptions{Name: "act"})
1097
s.NoError(w2.Start())
1098
defer w2.Stop()
1099
1100
env.WaitForChannel(timedoutTask)
1101
1102
// After scheduling the third time, now pending activity should be assigned to v3
1103
s.Eventually(
1104
func() bool {
1105
dw, err := env.SdkClient().DescribeWorkflowExecution(s.Context(), run.GetID(), run.GetRunID())
1106
s.NoError(err)
1107
s.Len(dw.GetPendingActivities(), 1)
1108
return v3 == dw.GetPendingActivities()[0].GetLastIndependentlyAssignedBuildId()
1109
},
1110
10*time.Second,
1111
50*time.Millisecond,
1112
)
1113
1114
act3 := func() (string, error) {
1115
return "done in v3!", nil
1116
}
1117
1118
// run worker on v3 so it can complete the wf
1119
w3 := worker.New(env.SdkClient(), actTq, worker.Options{
1120
BuildID: v3,
1121
UseBuildIDForVersioning: true,
1122
MaxConcurrentWorkflowTaskPollers: numPollers,
1123
})
1124
w3.RegisterActivityWithOptions(act3, activity.RegisterOptions{Name: "act"})
1125
s.NoError(w3.Start())
1126
defer w3.Stop()
1127
1128
var out string
1129
s.NoError(run.Get(s.Context(), &out))
1130
s.Equal("done!", out)
1131
1132
if versionedWf {
1133
s.validateWorkflowEventsVersionStamps(env,
1134
run.GetID(), run.GetRunID(), []string{
1135
wfV1,
1136
v3, // succeeded activity
1137
// wfV1, removed because it's on a sticky queue
1138
}, "",
1139
)
1140
} else {
1141
s.validateWorkflowEventsVersionStamps(env,
1142
run.GetID(), run.GetRunID(), []string{
1143
"",
1144
v3, // succeeded activity
1145
"",
1146
}, "",
1147
)
1148
}
1149
}
1150
1151
func (s *VersioningIntegSuite) TestIndependentActivityTaskAssignment_SyncMatch_VersionedWorkflow() {
1152
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) {
1153
s.independentActivityTaskAssignmentSyncMatch(env, true)
1154
})
1155
}
1156
1157
func (s *VersioningIntegSuite) TestIndependentActivityTaskAssignment_SyncMatch_UnversionedWorkflow() {
1158
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) {
1159
s.independentActivityTaskAssignmentSyncMatch(env, false)
1160
})
1161
}
1162
1163
func (s *VersioningIntegSuite) independentActivityTaskAssignmentSyncMatch(
1164
env *testcore.TestEnv,
1165
versionedWf bool,
1166
) {
1167
wfTq := testcore.RandomizeStr(s.T().Name())
1168
actTq := testcore.RandomizeStr(s.T().Name())
1169
wfV1 := s.prefixed("wfv1")
1170
v1 := s.prefixed("v1")
1171
v2 := s.prefixed("v2")
1172
v3 := s.prefixed("v3")
1173
1174
rule := s.addAssignmentRule(env, actTq, v1)
1175
s.waitForAssignmentRulePropagation(env, actTq, rule)
1176
if versionedWf {
1177
rule := s.addAssignmentRule(env, wfTq, wfV1)
1178
s.waitForAssignmentRulePropagation(env, wfTq, rule)
1179
}
1180
1181
wf := func(ctx workflow.Context) (string, error) {
1182
var ret string
1183
err := workflow.ExecuteActivity(workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
1184
TaskQueue: actTq,
1185
StartToCloseTimeout: 1 * time.Second,
1186
RetryPolicy: &temporal.RetryPolicy{
1187
InitialInterval: 1 * time.Second,
1188
BackoffCoefficient: 1,
1189
},
1190
}), "act").Get(ctx, &ret)
1191
s.NoError(err)
1192
s.Equal("done in v3!", ret)
1193
return "done!", nil
1194
}
1195
1196
// start workflow worker
1197
wfw := worker.New(env.SdkClient(), wfTq, worker.Options{
1198
BuildID: wfV1,
1199
UseBuildIDForVersioning: versionedWf,
1200
MaxConcurrentWorkflowTaskPollers: numPollers,
1201
})
1202
wfw.RegisterWorkflow(wf)
1203
s.NoError(wfw.Start())
1204
defer wfw.Stop()
1205
1206
// v1 fails the activity
1207
failedTask := make(chan struct{})
1208
act1 := func() (string, error) {
1209
close(failedTask)
1210
return "", errors.New("failing activity task intentionally")
1211
}
1212
1213
w1 := worker.New(env.SdkClient(), actTq, worker.Options{
1214
BuildID: v1,
1215
UseBuildIDForVersioning: true,
1216
MaxConcurrentWorkflowTaskPollers: numPollers,
1217
})
1218
w1.RegisterActivityWithOptions(act1, activity.RegisterOptions{Name: "act"})
1219
s.NoError(w1.Start())
1220
defer w1.Stop()
1221
1222
run, err := env.SdkClient().ExecuteWorkflow(s.Context(), sdkclient.StartWorkflowOptions{
1223
TaskQueue: wfTq,
1224
// setting to a short timeout to verify first wf task timeout assignment
1225
WorkflowTaskTimeout: 1 * time.Second,
1226
}, wf)
1227
s.NoError(err)
1228
1229
env.WaitForChannel(failedTask)
1230
1231
// MS should have the correct build ID after finishing the first WFT
1232
s.Eventually(
1233
func() bool {
1234
dw, err := env.SdkClient().DescribeWorkflowExecution(s.Context(), run.GetID(), run.GetRunID())
1235
s.NoError(err)
1236
if len(dw.GetPendingActivities()) == 0 {
1237
return false
1238
}
1239
if versionedWf {
1240
s.Equal(wfV1, dw.GetWorkflowExecutionInfo().GetAssignedBuildId())
1241
s.Equal(wfV1, dw.GetWorkflowExecutionInfo().GetMostRecentWorkerVersionStamp().GetBuildId())
1242
} else {
1243
s.Empty(dw.GetWorkflowExecutionInfo().GetAssignedBuildId()) //nolint:staticcheck
1244
s.False(dw.GetWorkflowExecutionInfo().GetMostRecentWorkerVersionStamp().GetUseVersioning())
1245
}
1246
return v1 == dw.GetPendingActivities()[0].GetLastIndependentlyAssignedBuildId()
1247
},
1248
10*time.Second,
1249
50*time.Millisecond,
1250
)
1251
1252
// v2 timesout the activity
1253
timedoutTask := make(chan struct{})
1254
act2 := func() (string, error) {
1255
time.Sleep(1 * time.Second) //nolint:forbidigo
1256
close(timedoutTask)
1257
time.Sleep(100 * time.Second) //nolint:forbidigo
1258
return "return after long sleep", nil
1259
}
1260
1261
w2 := worker.New(env.SdkClient(), actTq, worker.Options{
1262
BuildID: v2,
1263
UseBuildIDForVersioning: true,
1264
MaxConcurrentWorkflowTaskPollers: numPollers,
1265
// since the WF cache is shared by all workers in this process, we need to set this in a way that ensures
1266
// WFTs will timeout in w2 but not in w3
1267
DeadlockDetectionTimeout: 1500 * time.Millisecond,
1268
})
1269
w2.RegisterActivityWithOptions(act2, activity.RegisterOptions{Name: "act"})
1270
s.NoError(w2.Start())
1271
defer w2.Stop()
1272
1273
// update latest build to v2
1274
rule = s.addAssignmentRule(env, actTq, v2)
1275
s.waitForAssignmentRulePropagation(env, actTq, rule)
1276
1277
env.WaitForChannel(timedoutTask)
1278
1279
// After scheduling the second time, now pending activity should be assigned to v2
1280
s.Eventually(
1281
func() bool {
1282
dw, err := env.SdkClient().DescribeWorkflowExecution(s.Context(), run.GetID(), run.GetRunID())
1283
s.NoError(err)
1284
s.Len(dw.GetPendingActivities(), 1)
1285
return v2 == dw.GetPendingActivities()[0].GetLastIndependentlyAssignedBuildId()
1286
},
1287
10*time.Second,
1288
50*time.Millisecond,
1289
)
1290
1291
// v3 can process the activity
1292
act3 := func() (string, error) {
1293
return "done in v3!", nil
1294
}
1295
1296
// run worker on v3 so it can complete the wf
1297
w3 := worker.New(env.SdkClient(), actTq, worker.Options{
1298
BuildID: v3,
1299
UseBuildIDForVersioning: true,
1300
MaxConcurrentWorkflowTaskPollers: numPollers,
1301
})
1302
w3.RegisterActivityWithOptions(act3, activity.RegisterOptions{Name: "act"})
1303
s.NoError(w3.Start())
1304
defer w3.Stop()
1305
1306
// update latest build to v3
1307
rule = s.addAssignmentRule(env, actTq, v3)
1308
s.waitForAssignmentRulePropagation(env, actTq, rule)
1309
1310
var out string
1311
s.NoError(run.Get(s.Context(), &out))
1312
s.Equal("done!", out)
1313
1314
if versionedWf {
1315
s.validateWorkflowBuildIds(env, run.GetID(), run.GetRunID(), wfV1, true, wfV1, "", nil)
1316
s.validateWorkflowEventsVersionStamps(env,
1317
run.GetID(), run.GetRunID(), []string{
1318
wfV1,
1319
v3, // succeeded activity
1320
// wfV1, skipping stamp because this is a sticky queue task
1321
}, "",
1322
)
1323
} else {
1324
s.validateWorkflowBuildIds(env, run.GetID(), run.GetRunID(), "", true, wfV1, "", nil)
1325
s.validateWorkflowEventsVersionStamps(env,
1326
run.GetID(), run.GetRunID(), []string{
1327
"",
1328
v3, // succeeded activity
1329
"",
1330
}, "",
1331
)
1332
}
1333
}
1334
1335
func (s *VersioningIntegSuite) TestWorkflowTaskRedirectInRetryFirstTask() {
1336
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) {
1337
s.testWorkflowTaskRedirectInRetry(env, true)
1338
})
1339
}
1340
1341
func (s *VersioningIntegSuite) TestWorkflowTaskRedirectInRetryNonFirstTask() {
1342
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) {
1343
s.testWorkflowTaskRedirectInRetry(env, false)
1344
})
1345
}
1346
1347
func (s *VersioningIntegSuite) testWorkflowTaskRedirectInRetry(
1348
env *testcore.TestEnv,
1349
firstTask bool,
1350
) {
1351
tq := testcore.RandomizeStr(s.T().Name())
1352
v1 := s.prefixed("v1")
1353
v11 := s.prefixed("v1.1")
1354
v12 := s.prefixed("v1.2")
1355
1356
rule := s.addAssignmentRule(env, tq, v1)
1357
s.waitForAssignmentRulePropagation(env, tq, rule)
1358
1359
act := func() error {
1360
return nil
1361
}
1362
1363
// v1 fails the task
1364
failedTask := make(chan struct{})
1365
wf1 := func(ctx workflow.Context) (string, error) {
1366
if !firstTask {
1367
// add an activity to move workflow past first WFT
1368
var out any
1369
err := workflow.ExecuteActivity(workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
1370
StartToCloseTimeout: 1 * time.Second}), act).Get(ctx, &out)
1371
s.NoError(err)
1372
}
1373
failedTask <- struct{}{}
1374
panic("failing WF task intentionally")
1375
}
1376
1377
w1 := worker.New(env.SdkClient(), tq, worker.Options{
1378
BuildID: v1,
1379
UseBuildIDForVersioning: true,
1380
MaxConcurrentWorkflowTaskPollers: numPollers,
1381
})
1382
w1.RegisterWorkflowWithOptions(wf1, workflow.RegisterOptions{Name: "wf"})
1383
w1.RegisterActivity(act)
1384
s.NoError(w1.Start())
1385
defer w1.Stop()
1386
1387
run, err := env.SdkClient().ExecuteWorkflow(s.Context(), sdkclient.StartWorkflowOptions{
1388
TaskQueue: tq,
1389
// setting to a short timeout to verify first wf task timeout assignment
1390
WorkflowTaskTimeout: 1 * time.Second,
1391
RetryPolicy: &temporal.RetryPolicy{
1392
BackoffCoefficient: 1.,
1393
InitialInterval: time.Second,
1394
},
1395
}, "wf")
1396
s.NoError(err)
1397
1398
// wait for multiple failures to make sure more attempts does not generate more history tasks
1399
env.WaitForChannel(failedTask)
1400
env.WaitForChannel(failedTask)
1401
env.WaitForChannel(failedTask)
1402
1403
expectedStampBuildId := ""
1404
if !firstTask {
1405
expectedStampBuildId = v1
1406
}
1407
// MS should have the correct build ID
1408
s.validateWorkflowBuildIds(env, run.GetID(), run.GetRunID(), v1, true, expectedStampBuildId, "", nil)
1409
1410
// v11 times out the task
1411
timedoutTask := make(chan struct{})
1412
wf11 := func(ctx workflow.Context) (string, error) {
1413
if !firstTask {
1414
// add an activity to move workflow past first WFT
1415
var out any
1416
err := workflow.ExecuteActivity(workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
1417
StartToCloseTimeout: 1 * time.Second}), act).Get(ctx, &out)
1418
s.NoError(err)
1419
}
1420
time.Sleep(1 * time.Second) //nolint:forbidigo
1421
timedoutTask <- struct{}{}
1422
time.Sleep(100 * time.Second) //nolint:forbidigo
1423
return "return after long sleep", nil
1424
}
1425
1426
w11 := worker.New(env.SdkClient(), tq, worker.Options{
1427
BuildID: v11,
1428
UseBuildIDForVersioning: true,
1429
MaxConcurrentWorkflowTaskPollers: numPollers,
1430
// since the WF cache is shared by all workers in this process, we need to set this in a way that ensures
1431
// WFTs will timeout in w2 but not in w3
1432
DeadlockDetectionTimeout: 1500 * time.Millisecond,
1433
})
1434
w11.RegisterWorkflowWithOptions(wf11, workflow.RegisterOptions{Name: "wf"})
1435
w11.RegisterActivity(act)
1436
s.NoError(w11.Start())
1437
defer w11.Stop()
1438
1439
rule2 := s.addRedirectRule(env, tq, v1, v11)
1440
s.waitForRedirectRulePropagation(env, tq, rule2)
1441
1442
// wait for multiple timeouts to make sure more attempts does not generate more history tasks
1443
env.WaitForChannel(timedoutTask)
1444
env.WaitForChannel(timedoutTask)
1445
env.WaitForChannel(timedoutTask)
1446
// After scheduling the second time, now MS should be assigned to v2
1447
s.validateWorkflowBuildIds(env, run.GetID(), run.GetRunID(), v11, true, expectedStampBuildId, "", []string{v1})
1448
1449
// v12 can process the task
1450
wf12 := func(ctx workflow.Context) (string, error) {
1451
if !firstTask {
1452
// add an activity to move workflow past first WFT
1453
var out any
1454
err := workflow.ExecuteActivity(workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
1455
StartToCloseTimeout: 1 * time.Second}), act).Get(ctx, &out)
1456
s.NoError(err)
1457
}
1458
return "done on v1.2!", nil
1459
}
1460
1461
// run worker on v12 so it can complete the wf
1462
w12 := worker.New(env.SdkClient(), tq, worker.Options{
1463
BuildID: v12,
1464
UseBuildIDForVersioning: true,
1465
MaxConcurrentWorkflowTaskPollers: numPollers,
1466
})
1467
w12.RegisterWorkflowWithOptions(wf12, workflow.RegisterOptions{Name: "wf"})
1468
w12.RegisterActivity(act)
1469
s.NoError(w12.Start())
1470
defer w12.Stop()
1471
1472
rule2 = s.addRedirectRule(env, tq, v11, v12)
1473
s.waitForRedirectRulePropagation(env, tq, rule2)
1474
1475
var out string
1476
s.NoError(run.Get(s.Context(), &out))
1477
s.Equal("done on v1.2!", out)
1478
s.validateWorkflowBuildIds(env, run.GetID(), run.GetRunID(), v12, true, v12, "", []string{v1, v11})
1479
expectedStamps := []string{
1480
v1, // failed wf task
1481
v11, // timed out wf task show up in history because a redirect happened after them
1482
v12, // succeeded wf task
1483
}
1484
if !firstTask {
1485
expectedStamps = []string{
1486
v1, // first wf task
1487
v1, // activity task
1488
// v1, // skipping stamp for failed wf task on sticky queue
1489
v1, // failed wf task on normal queue
1490
v11, // timed out wf task show up in history because they happened on a different build ID
1491
v12, // succeeded wf task
1492
}
1493
}
1494
s.validateWorkflowEventsVersionStamps(env, run.GetID(), run.GetRunID(), expectedStamps, "")
1495
}
1496
1497
func (s *VersioningIntegSuite) TestDispatchNotUsingVersioningOld() {
1498
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) { s.dispatchNotUsingVersioning(env, false) })
1499
}
1500
1501
func (s *VersioningIntegSuite) TestDispatchNotUsingVersioning() {
1502
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) { s.dispatchNotUsingVersioning(env, true) })
1503
}
1504
1505
func (s *VersioningIntegSuite) dispatchNotUsingVersioning(
1506
env *testcore.TestEnv,
1507
newVersioning bool,
1508
) {
1509
tq := testcore.RandomizeStr(s.T().Name())
1510
v1 := s.prefixed("v1")
1511
1512
wf1nover := func(ctx workflow.Context) (string, error) {
1513
return "done without versioning!", nil
1514
}
1515
wf1 := func(ctx workflow.Context) (string, error) {
1516
return "done with versioning!", nil
1517
}
1518
1519
if newVersioning {
1520
rule := s.addAssignmentRule(env, tq, v1)
1521
s.waitForAssignmentRulePropagation(env, tq, rule)
1522
} else {
1523
s.addNewDefaultBuildID(env, tq, v1)
1524
s.waitForVersionSetPropagation(env, tq, v1)
1525
}
1526
1527
w1nover := worker.New(env.SdkClient(), tq, worker.Options{
1528
BuildID: v1,
1529
UseBuildIDForVersioning: false,
1530
MaxConcurrentWorkflowTaskPollers: numPollers,
1531
})
1532
w1 := worker.New(env.SdkClient(), tq, worker.Options{
1533
BuildID: v1,
1534
UseBuildIDForVersioning: true,
1535
MaxConcurrentWorkflowTaskPollers: numPollers,
1536
})
1537
w1nover.RegisterWorkflowWithOptions(wf1nover, workflow.RegisterOptions{Name: "wf"})
1538
w1.RegisterWorkflowWithOptions(wf1, workflow.RegisterOptions{Name: "wf"})
1539
s.NoError(w1nover.Start())
1540
defer w1nover.Stop()
1541
s.NoError(w1.Start())
1542
defer w1.Stop()
1543
1544
run, err := env.SdkClient().ExecuteWorkflow(s.Context(), sdkclient.StartWorkflowOptions{TaskQueue: tq}, "wf")
1545
s.NoError(err)
1546
var out string
1547
s.NoError(run.Get(s.Context(), &out))
1548
s.Equal("done with versioning!", out)
1549
1550
s.validateWorkflowBuildIds(env, run.GetID(), run.GetRunID(), v1, newVersioning, v1, "", nil)
1551
}
1552
1553
func (s *VersioningIntegSuite) TestDispatchNewWorkflowStartWorkerFirst() {
1554
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) { s.dispatchNewWorkflowStartWorkerFirst(env) })
1555
}
1556
1557
func (s *VersioningIntegSuite) dispatchNewWorkflowStartWorkerFirst(env *testcore.TestEnv) {
1558
tq := testcore.RandomizeStr(s.T().Name())
1559
v1 := s.prefixed("v1")
1560
1561
wf := func(ctx workflow.Context) (string, error) {
1562
return "done!", nil
1563
}
1564
1565
// run worker before registering build. it will use guessed set id
1566
w1 := worker.New(env.SdkClient(), tq, worker.Options{
1567
BuildID: v1,
1568
UseBuildIDForVersioning: true,
1569
MaxConcurrentWorkflowTaskPollers: numPollers,
1570
})
1571
w1.RegisterWorkflow(wf)
1572
s.NoError(w1.Start())
1573
defer w1.Stop()
1574
1575
// wait for it to start polling
1576
time.Sleep(200 * time.Millisecond) //nolint:forbidigo
1577
1578
s.addNewDefaultBuildID(env, tq, v1)
1579
s.waitForVersionSetPropagation(env, tq, v1)
1580
1581
run, err := env.SdkClient().ExecuteWorkflow(s.Context(), sdkclient.StartWorkflowOptions{TaskQueue: tq}, wf)
1582
s.NoError(err)
1583
var out string
1584
s.NoError(run.Get(s.Context(), &out))
1585
s.Equal("done!", out)
1586
1587
s.validateWorkflowBuildIds(env, run.GetID(), run.GetRunID(), v1, false, v1, "", nil)
1588
}
1589
1590
func (s *VersioningIntegSuite) TestDispatchUnversionedRemainsUnversioned() {
1591
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) {
1592
s.dispatchUnversionedRemainsUnversioned(env)
1593
})
1594
}
1595
1596
func (s *VersioningIntegSuite) dispatchUnversionedRemainsUnversioned(env *testcore.TestEnv) {
1597
tq := testcore.RandomizeStr(s.T().Name())
1598
v1 := s.prefixed("v1")
1599
1600
started := make(chan struct{}, 1)
1601
1602
wf := func(ctx workflow.Context) (string, error) {
1603
started <- struct{}{}
1604
workflow.GetSignalChannel(ctx, "wait").Receive(ctx, nil)
1605
return "done!", nil
1606
}
1607
1608
w1 := worker.New(env.SdkClient(), tq, worker.Options{
1609
// no build ID
1610
})
1611
w1.RegisterWorkflow(wf)
1612
s.NoError(w1.Start())
1613
defer w1.Stop()
1614
1615
run, err := env.SdkClient().ExecuteWorkflow(s.Context(), sdkclient.StartWorkflowOptions{TaskQueue: tq}, wf)
1616
s.NoError(err)
1617
1618
env.WaitForChannel(started)
1619
s.addNewDefaultBuildID(env, tq, v1)
1620
s.waitForVersionSetPropagation(env, tq, v1)
1621
1622
// unblock the workflow
1623
s.NoError(env.SdkClient().SignalWorkflow(s.Context(), run.GetID(), run.GetRunID(), "wait", nil))
1624
1625
var out string
1626
s.NoError(run.Get(s.Context(), &out))
1627
s.Equal("done!", out)
1628
1629
s.validateWorkflowBuildIds(env, run.GetID(), run.GetRunID(), "", false, "binary-checksum", "", nil)
1630
}
1631
1632
func (s *VersioningIntegSuite) TestDispatchUpgradeStopOldOld() {
1633
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) { s.dispatchUpgrade(env, false, true) })
1634
}
1635
1636
func (s *VersioningIntegSuite) TestDispatchUpgradeWaitOld() {
1637
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) { s.dispatchUpgrade(env, false, false) })
1638
}
1639
1640
func (s *VersioningIntegSuite) TestDispatchUpgradeStopOld() {
1641
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) { s.dispatchUpgrade(env, true, true) })
1642
}
1643
1644
func (s *VersioningIntegSuite) TestDispatchUpgradeWait() {
1645
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) { s.dispatchUpgrade(env, true, false) })
1646
}
1647
1648
func (s *VersioningIntegSuite) dispatchUpgrade(
1649
env *testcore.TestEnv,
1650
newVersioning, stopOld bool,
1651
) {
1652
tq := testcore.RandomizeStr(s.T().Name())
1653
v1 := s.prefixed("v1")
1654
v11 := s.prefixed("v11")
1655
1656
started := make(chan struct{}, 1)
1657
1658
wf1 := func(ctx workflow.Context) (string, error) {
1659
started <- struct{}{}
1660
workflow.GetSignalChannel(ctx, "wait").Receive(ctx, nil)
1661
return "done!", nil
1662
}
1663
1664
wf11 := func(ctx workflow.Context) (string, error) {
1665
workflow.GetSignalChannel(ctx, "wait").Receive(ctx, nil)
1666
return "done from 1.1!", nil
1667
}
1668
1669
if newVersioning {
1670
rule := s.addAssignmentRule(env, tq, v1)
1671
s.waitForAssignmentRulePropagation(env, tq, rule)
1672
} else {
1673
s.addNewDefaultBuildID(env, tq, v1)
1674
s.waitForVersionSetPropagation(env, tq, v1)
1675
}
1676
1677
w1 := worker.New(env.SdkClient(), tq, worker.Options{
1678
BuildID: v1,
1679
UseBuildIDForVersioning: true,
1680
MaxConcurrentWorkflowTaskPollers: numPollers,
1681
})
1682
w1.RegisterWorkflowWithOptions(wf1, workflow.RegisterOptions{Name: "wf"})
1683
s.NoError(w1.Start())
1684
defer w1.Stop()
1685
1686
run, err := env.SdkClient().ExecuteWorkflow(s.Context(), sdkclient.StartWorkflowOptions{TaskQueue: tq}, "wf")
1687
s.NoError(err)
1688
env.WaitForChannel(started)
1689
1690
// now add v11 as compatible so the next workflow task runs there
1691
if newVersioning {
1692
rule := s.addRedirectRule(env, tq, v1, v11)
1693
s.waitForRedirectRulePropagation(env, tq, rule)
1694
} else {
1695
s.addCompatibleBuildID(env, tq, v11, v1, false)
1696
s.waitForVersionSetPropagation(env, tq, v11)
1697
}
1698
// add another 100ms to make sure it got to sticky queues also
1699
time.Sleep(100 * time.Millisecond) //nolint:forbidigo
1700
1701
w11 := worker.New(env.SdkClient(), tq, worker.Options{
1702
BuildID: v11,
1703
UseBuildIDForVersioning: true,
1704
MaxConcurrentWorkflowTaskPollers: numPollers,
1705
})
1706
w11.RegisterWorkflowWithOptions(wf11, workflow.RegisterOptions{Name: "wf"})
1707
s.NoError(w11.Start())
1708
defer w11.Stop()
1709
1710
// Two cases:
1711
if stopOld {
1712
// Stop the old worker. Workflow tasks will go to the sticky queue, which will see that
1713
// it's not the latest and kick them back to the normal queue, which will be dispatched
1714
// to v11.
1715
w1.Stop()
1716
} else {
1717
// Don't stop the old worker.
1718
if !newVersioning {
1719
// In this case, w1 will still have some pollers blocked on
1720
// the normal queue which could pick up tasks that we want to go to v11. (We don't
1721
// interrupt long polls.) To ensure those polls don't interfere, wait for them to
1722
// expire.
1723
time.Sleep(longPollTime) //nolint:forbidigo
1724
}
1725
}
1726
1727
// unblock the workflow
1728
s.NoError(env.SdkClient().SignalWorkflow(s.Context(), run.GetID(), run.GetRunID(), "wait", nil))
1729
1730
var out string
1731
s.NoError(run.Get(s.Context(), &out))
1732
s.Equal("done from 1.1!", out)
1733
1734
s.validateWorkflowBuildIds(env, run.GetID(), run.GetRunID(), v11, newVersioning, v11, "", []string{v1})
1735
}
1736
1737
type activityFailMode int
1738
1739
const (
1740
dontFailActivity = iota
1741
failActivity
1742
timeoutActivity
1743
)
1744
1745
func (s *VersioningIntegSuite) TestDispatchActivityOld() {
1746
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) {
1747
s.dispatchActivity(env, dontFailActivity, false, false)
1748
})
1749
}
1750
1751
func (s *VersioningIntegSuite) TestDispatchActivityFailOld() {
1752
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) {
1753
s.dispatchActivity(env, failActivity, false, false)
1754
})
1755
}
1756
1757
func (s *VersioningIntegSuite) TestDispatchActivityTimeoutOld() {
1758
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) {
1759
s.dispatchActivity(env, timeoutActivity, false, false)
1760
})
1761
}
1762
1763
func (s *VersioningIntegSuite) TestDispatchActivity() {
1764
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) {
1765
s.dispatchActivity(env, dontFailActivity, true, false)
1766
})
1767
}
1768
1769
func (s *VersioningIntegSuite) TestDispatchActivityFail() {
1770
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) {
1771
s.dispatchActivity(env, failActivity, true, false)
1772
})
1773
}
1774
1775
func (s *VersioningIntegSuite) TestDispatchActivityTimeout() {
1776
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) {
1777
s.dispatchActivity(env, timeoutActivity, true, false)
1778
})
1779
}
1780
1781
func (s *VersioningIntegSuite) TestDispatchActivityCrossTq() {
1782
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) {
1783
s.dispatchActivity(env, dontFailActivity, true, true)
1784
})
1785
}
1786
1787
func (s *VersioningIntegSuite) TestDispatchActivityFailCrossTq() {
1788
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) {
1789
s.dispatchActivity(env, failActivity, true, true)
1790
})
1791
}
1792
1793
func (s *VersioningIntegSuite) TestDispatchActivityTimeoutCrossTq() {
1794
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) {
1795
s.dispatchActivity(env, timeoutActivity, true, true)
1796
})
1797
}
1798
1799
func (s *VersioningIntegSuite) dispatchActivity(
1800
env *testcore.TestEnv,
1801
failMode activityFailMode, newVersioning bool, crossTq bool,
1802
) {
1803
// This also implicitly tests that a workflow stays on a compatible version set if a new
1804
// incompatible set is registered, because wf2 just panics. It further tests that
1805
// stickiness on v1 is not broken by registering v2, because the channel send will panic on
1806
// replay after we close the channel.
1807
1808
tq := testcore.RandomizeStr(s.T().Name())
1809
actxTq := testcore.RandomizeStr(s.T().Name() + "activity")
1810
v1 := s.prefixed("v1")
1811
v2 := s.prefixed("v2")
1812
1813
started := make(chan struct{}, 1)
1814
1815
var act1state, act2state atomic.Int32
1816
1817
doAct := func(state *atomic.Int32, output string) (string, error) {
1818
if state.Add(1) == 1 {
1819
switch failMode {
1820
case failActivity:
1821
return "", errors.New("try again")
1822
case timeoutActivity:
1823
time.Sleep(5 * time.Second) //nolint:forbidigo
1824
return "ignored", nil
1825
}
1826
}
1827
return output, nil
1828
}
1829
1830
act1 := func() (string, error) {
1831
return doAct(&act1state, "v1")
1832
}
1833
act2 := func() (string, error) {
1834
return doAct(&act2state, "v2")
1835
}
1836
act1xTq := func() (string, error) {
1837
return doAct(&act1state, "v1xTq")
1838
}
1839
act2xTq := func() (string, error) {
1840
return doAct(&act2state, "v2xTq")
1841
}
1842
wf1 := func(ctx workflow.Context) (string, error) {
1843
started <- struct{}{}
1844
// wait for signal
1845
workflow.GetSignalChannel(ctx, "wait").Receive(ctx, nil)
1846
actTq := tq
1847
if crossTq {
1848
actTq = actxTq
1849
}
1850
// run two activities
1851
fut1 := workflow.ExecuteActivity(workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
1852
ScheduleToCloseTimeout: time.Minute,
1853
DisableEagerExecution: true,
1854
VersioningIntent: temporal.VersioningIntentCompatible,
1855
StartToCloseTimeout: 1 * time.Second,
1856
TaskQueue: actTq,
1857
}), "act")
1858
fut2 := workflow.ExecuteActivity(workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
1859
ScheduleToCloseTimeout: time.Minute,
1860
DisableEagerExecution: true,
1861
VersioningIntent: temporal.VersioningIntentDefault, // this one should go to default
1862
StartToCloseTimeout: 1 * time.Second,
1863
TaskQueue: actTq,
1864
}), "act")
1865
var val1, val2 string
1866
s.NoError(fut1.Get(ctx, &val1))
1867
s.NoError(fut2.Get(ctx, &val2))
1868
return val1 + val2, nil
1869
}
1870
wf2 := func(ctx workflow.Context) (string, error) {
1871
panic("workflow should not run on v2")
1872
}
1873
1874
if newVersioning {
1875
rule := s.addAssignmentRule(env, tq, v1)
1876
s.waitForAssignmentRulePropagation(env, tq, rule)
1877
} else {
1878
s.addNewDefaultBuildID(env, tq, v1)
1879
s.waitForVersionSetPropagation(env, tq, v1)
1880
}
1881
1882
w1 := worker.New(env.SdkClient(), tq, worker.Options{
1883
BuildID: v1,
1884
UseBuildIDForVersioning: true,
1885
MaxConcurrentWorkflowTaskPollers: numPollers,
1886
})
1887
w1.RegisterWorkflowWithOptions(wf1, workflow.RegisterOptions{Name: "wf"})
1888
w1.RegisterActivityWithOptions(act1, activity.RegisterOptions{Name: "act"})
1889
s.NoError(w1.Start())
1890
defer w1.Stop()
1891
w1xTq := worker.New(env.SdkClient(), actxTq, worker.Options{
1892
BuildID: v1,
1893
UseBuildIDForVersioning: true,
1894
MaxConcurrentWorkflowTaskPollers: numPollers,
1895
})
1896
w1xTq.RegisterActivityWithOptions(act1xTq, activity.RegisterOptions{Name: "act"})
1897
s.NoError(w1xTq.Start())
1898
defer w1xTq.Stop()
1899
1900
run, err := env.SdkClient().ExecuteWorkflow(s.Context(), sdkclient.StartWorkflowOptions{TaskQueue: tq}, "wf")
1901
s.NoError(err)
1902
// wait for it to start on v1
1903
env.WaitForChannel(started)
1904
close(started) // force panic if replayed
1905
1906
// now register v2 as default
1907
if newVersioning {
1908
rule := s.addAssignmentRule(env, tq, v2)
1909
s.waitForAssignmentRulePropagation(env, tq, rule)
1910
rule = s.addAssignmentRule(env, actxTq, v2)
1911
s.waitForAssignmentRulePropagation(env, actxTq, rule)
1912
} else {
1913
s.addNewDefaultBuildID(env, tq, v2)
1914
s.waitForVersionSetPropagation(env, tq, v2)
1915
}
1916
1917
// start worker for v2
1918
w2 := worker.New(env.SdkClient(), tq, worker.Options{
1919
BuildID: v2,
1920
UseBuildIDForVersioning: true,
1921
MaxConcurrentWorkflowTaskPollers: numPollers,
1922
})
1923
w2.RegisterWorkflowWithOptions(wf2, workflow.RegisterOptions{Name: "wf"})
1924
w2.RegisterActivityWithOptions(act2, activity.RegisterOptions{Name: "act"})
1925
s.NoError(w2.Start())
1926
defer w2.Stop()
1927
w2xTq := worker.New(env.SdkClient(), actxTq, worker.Options{
1928
BuildID: v2,
1929
UseBuildIDForVersioning: true,
1930
MaxConcurrentWorkflowTaskPollers: numPollers,
1931
})
1932
w2xTq.RegisterActivityWithOptions(act2xTq, activity.RegisterOptions{Name: "act"})
1933
s.NoError(w2xTq.Start())
1934
defer w2xTq.Stop()
1935
1936
// unblock the workflow
1937
s.NoError(env.SdkClient().SignalWorkflow(s.Context(), run.GetID(), run.GetRunID(), "wait", nil))
1938
1939
var out string
1940
s.NoError(run.Get(s.Context(), &out))
1941
if crossTq {
1942
s.Equal("v1xTqv2xTq", out)
1943
} else {
1944
s.Equal("v1v2", out)
1945
}
1946
1947
s.validateWorkflowBuildIds(env, run.GetID(), run.GetRunID(), v1, newVersioning, v1, "", nil)
1948
}
1949
1950
func (s *VersioningIntegSuite) TestDispatchActivityUpgrade() {
1951
env := s.setupEnv()
1952
// Testing:
1953
// - redirecting dependent activity redirects wf
1954
// - this also implicitly verifies that sticky queue is cleared after redirecting activity task
1955
// - redirecting independent activity does not redirect wf
1956
1957
tq := testcore.RandomizeStr(s.T().Name())
1958
v1 := s.prefixed("v1")
1959
v11 := s.prefixed("v1.1")
1960
v12 := s.prefixed("v1.2")
1961
1962
startedWf := make(chan struct{}, 1)
1963
proceedWf := make(chan struct{}, 1)
1964
started11 := make(chan struct{}, 1)
1965
proceed11 := make(chan struct{}, 1)
1966
started12 := make(chan struct{}, 1)
1967
proceed12 := make(chan struct{}, 1)
1968
1969
act1 := func() (string, error) {
1970
return "v1", nil
1971
}
1972
act11 := func() (string, error) {
1973
started11 <- struct{}{}
1974
env.WaitForChannel(proceed11)
1975
return "v1.1", nil
1976
}
1977
act12 := func() (string, error) {
1978
started12 <- struct{}{}
1979
env.WaitForChannel(proceed12)
1980
return "v1.2", nil
1981
}
1982
wf := func(ctx workflow.Context) (string, error) {
1983
startedWf <- struct{}{}
1984
// Wait for adding v1 -> v1.1 redirect rule. We use a channel instead of WF signal to make sure the rules are
1985
// added between WFT started and completed and next task seeing the updated rules is an activity, not a WFT.
1986
<-proceedWf
1987
1988
var val1, val2 string
1989
// run two activities
1990
err1 := workflow.ExecuteActivity(workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
1991
ScheduleToCloseTimeout: time.Minute,
1992
DisableEagerExecution: true,
1993
VersioningIntent: temporal.VersioningIntentCompatible,
1994
StartToCloseTimeout: 1 * time.Second,
1995
}), "act").Get(ctx, &val1)
1996
s.NoError(err1)
1997
1998
startedWf <- struct{}{}
1999
// Wait for adding v1.1 -> v1.2 redirect rule. We use a channel instead of WF signal to make sure the rules are
2000
// added between WFT started and completed and next task seeing the updated rules is an activity, not a WFT.
2001
<-proceedWf
2002
2003
err2 := workflow.ExecuteActivity(workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
2004
ScheduleToCloseTimeout: time.Minute,
2005
DisableEagerExecution: true,
2006
VersioningIntent: temporal.VersioningIntentDefault,
2007
StartToCloseTimeout: 1 * time.Second,
2008
}), "act").Get(ctx, &val2)
2009
s.NoError(err2)
2010
2011
return val1 + val2, nil
2012
}
2013
2014
w1 := worker.New(env.SdkClient(), tq, worker.Options{
2015
BuildID: v1,
2016
UseBuildIDForVersioning: true,
2017
MaxConcurrentWorkflowTaskPollers: numPollers,
2018
DeadlockDetectionTimeout: 5 * time.Second,
2019
})
2020
w1.RegisterWorkflowWithOptions(wf, workflow.RegisterOptions{Name: "wf"})
2021
w1.RegisterActivityWithOptions(act1, activity.RegisterOptions{Name: "act"})
2022
s.NoError(w1.Start())
2023
defer w1.Stop()
2024
2025
w11 := worker.New(env.SdkClient(), tq, worker.Options{
2026
BuildID: v11,
2027
UseBuildIDForVersioning: true,
2028
MaxConcurrentWorkflowTaskPollers: numPollers,
2029
DeadlockDetectionTimeout: 5 * time.Second,
2030
})
2031
w11.RegisterWorkflowWithOptions(wf, workflow.RegisterOptions{Name: "wf"})
2032
w11.RegisterActivityWithOptions(act11, activity.RegisterOptions{Name: "act"})
2033
s.NoError(w11.Start())
2034
defer w11.Stop()
2035
2036
w12 := worker.New(env.SdkClient(), tq, worker.Options{
2037
BuildID: v12,
2038
UseBuildIDForVersioning: true,
2039
MaxConcurrentWorkflowTaskPollers: numPollers,
2040
DeadlockDetectionTimeout: 5 * time.Second,
2041
})
2042
w12.RegisterWorkflowWithOptions(wf, workflow.RegisterOptions{Name: "wf"})
2043
w12.RegisterActivityWithOptions(act12, activity.RegisterOptions{Name: "act"})
2044
s.NoError(w12.Start())
2045
defer w12.Stop()
2046
2047
rule := s.addAssignmentRule(env, tq, v1)
2048
s.waitForAssignmentRulePropagation(env, tq, rule)
2049
2050
run, err := env.SdkClient().ExecuteWorkflow(s.Context(), sdkclient.StartWorkflowOptions{TaskQueue: tq}, "wf")
2051
s.NoError(err)
2052
2053
// wait for it to start on v1
2054
env.WaitForChannel(startedWf)
2055
rule2 := s.addRedirectRule(env, tq, v1, v11)
2056
s.waitForRedirectRulePropagation(env, tq, rule2)
2057
env.SendToChannel(proceedWf)
2058
2059
env.WaitForChannel(started11)
2060
// wf assigned build ID should be updated by activity redirect
2061
s.validateWorkflowBuildIds(env, run.GetID(), run.GetRunID(), v11, true, v1, "", []string{v1})
2062
// let activity finish
2063
env.SendToChannel(proceed11)
2064
2065
// wf replays on 1.1 so need to unblock it an extra time
2066
env.WaitForChannel(startedWf)
2067
env.SendToChannel(proceedWf)
2068
2069
env.WaitForChannel(startedWf)
2070
rule2 = s.addRedirectRule(env, tq, v11, v12)
2071
s.waitForRedirectRulePropagation(env, tq, rule2)
2072
env.SendToChannel(proceedWf)
2073
2074
env.WaitForChannel(started12)
2075
// wf assigned build ID should not be updated by independent activity redirect
2076
s.validateWorkflowBuildIds(env, run.GetID(), run.GetRunID(), v11, true, v11, "", []string{v1})
2077
// let activity finish
2078
env.SendToChannel(proceed12)
2079
2080
// wf replays on 1.2 so need to unblock it two extra times
2081
env.WaitForChannel(startedWf)
2082
env.SendToChannel(proceedWf)
2083
env.WaitForChannel(startedWf)
2084
env.SendToChannel(proceedWf)
2085
2086
var out string
2087
s.NoError(run.Get(s.Context(), &out))
2088
s.Equal("v1.1v1.2", out)
2089
2090
s.validateWorkflowBuildIds(env, run.GetID(), run.GetRunID(), v12, true, v12, "", []string{v1, v11})
2091
}
2092
2093
func (s *VersioningIntegSuite) TestRedirectWithConcurrentActivities() {
2094
s.T().Skip("Skipping test since this tests old versioning behavior and also flakes")
2095
env := s.setupEnv(
2096
testcore.WithDynamicConfig(dynamicconfig.MatchingGetUserDataLongPollTimeout, 2*time.Second),
2097
)
2098
// Testing that wf never "goes back" to older build ID in presence of concurrent activities and random failures.
2099
//
2100
// SETUP:
2101
// 1- Run workers with build IDs v1.0 to v1.9.
2102
// 2- Workflows runs a nested loop of 10 * `activityRuns`. First loop is per build ID, second is to run parallel
2103
// activities for each build ID.
2104
// 3- Activities fail by `activityErrorRate` probability. Otherwise, they return success after a random wait.
2105
// 4- The first activity seeing a build ID (i.e. being run by that build ID) removes the redirect rule targeting
2106
// that build ID to test the workflow behaves correctly in absence of the applied redirect rule.
2107
// 5- One activity in each batch is responsible to add redirect rule to the next build ID. since, there is random
2108
// delay in activities, more activities of the same batch likely start both before and after this activity.
2109
// 6- Workflow waits for completion of all activities in a batch before going to the next one.
2110
//
2111
// VERIFYING:
2112
// 1- After each activity is completed, in workflow we verify that it is not completed by a build ID newer of that
2113
// of the workflow worker.
2114
// 2- Workflow finishes. I.e. no task is dropped without rescheduling due to build ID mismatch.
2115
// 3- At the end of the test we verify that the wf and activity started event build IDs are compatible with
2116
// workflows output (which contains the build ID of each completed activity)
2117
// 4- Redirect counter of all started events is valid
2118
// 5- Redirect rules were applied at least to one activity.
2119
// 6- At least one activity was retried.
2120
// 7- Some history events are unordered based on event timestamp (due to parallel activity random delay)
2121
2122
// Reduce user data long poll time for faster propagation of the versioning data. This is needed because of the
2123
// exponential minWaitTime logic in userDataManagerImpl that gets triggered because rules change very fast in
2124
// this test.
2125
2126
tq := testcore.RandomizeStr(s.T().Name())
2127
v1 := s.prefixed("v1.0")
2128
2129
rule := s.addAssignmentRule(env, tq, v1)
2130
s.waitForAssignmentRulePropagation(env, tq, rule)
2131
2132
versions := []string{v1}
2133
for v := 1; v <= 9; v++ {
2134
versions = append(versions, s.prefixed("v1."+strconv.Itoa(v)))
2135
}
2136
2137
activityErrorRate := .2 // chance of each activity attempt fail
2138
activityRuns := int32(10) // run the activity ~10 times on each version
2139
2140
activityCounter := atomic.Int32{}
2141
triggerRedirectAtActivityRun := atomic.Int32{}
2142
triggerRedirectAtActivityRun.Store(1)
2143
2144
lastRedirectTarget := atomic.Value{}
2145
lastRedirectTarget.Store(versions[0] + " redirect cleaned")
2146
var workers []worker.Worker
2147
2148
act := func(version string, runID int32) (string, error) {
2149
runs := activityCounter.Add(1)
2150
s.T().Logf("Starting activity %d on %s at %d\n", runID, version, runs)
2151
if lastRedirectTarget.CompareAndSwap(version, version+" observed") && version != versions[0] {
2152
// The last redirect rule is applied and observed by an activity, now delete it to make sure wf keeps using
2153
// the right build ID after applying the redirect rule, even when the rule is not present anymore.
2154
index, err := strconv.Atoi(version[len(version)-1:]) // get the last char of version is the index in the versions array
2155
s.NoError(err)
2156
s.T().Logf("Removing redirect from %s to %s \n", versions[index-1], version)
2157
s.removeRedirectRule(env, tq, versions[index-1])
2158
lastRedirectTarget.CompareAndSwap(version+" observed", version+" redirect cleaned")
2159
}
2160
if rand.Float64() < activityErrorRate {
2161
return "", errors.New("intentionally failing activity")
2162
}
2163
if triggerRedirectAtActivityRun.Load() == runID {
2164
// When enough activities are run using the current version, add redirect rule to the next version.
2165
v := runID / activityRuns
2166
if int(v+1) < len(versions) {
2167
// wait for last redirect rule to be cleaned up
2168
for !lastRedirectTarget.CompareAndSwap(versions[v]+" redirect cleaned", versions[v+1]) {
2169
}
2170
s.T().Logf("Adding redirect from %s to %s at %d\n", versions[v], versions[v+1], runs)
2171
s.addRedirectRule(env, tq, versions[v], versions[v+1])
2172
// Intentionally do not wait for propagation of the rules to partitions. Waiting will linger this
2173
// activity and allows all the other concurrent activities to finish, leaving only the WFT task to
2174
// see the redirect rule for the first time.
2175
triggerRedirectAtActivityRun.CompareAndSwap(runID, runID+activityRuns)
2176
}
2177
}
2178
2179
// Add random sleep to simulate network delay
2180
//nolint:forbidigo
2181
time.Sleep(time.Duration(int64(rand.Intn(50)) * int64(time.Millisecond))) //nolint:forbidigo
2182
s.T().Logf("Completing activity %d on %s at %d\n", runID, version, runs)
2183
return version, nil
2184
}
2185
2186
wf := func(wfCtx workflow.Context, wfVersion string) (string, error) {
2187
var res []string
2188
// because of rule propagation delay we run for more than 10 cycles to make sure all versions are seen
2189
for i := 0; i <= 12; i++ {
2190
var futures []workflow.Future
2191
for j := 0; j < int(activityRuns); j++ {
2192
f := workflow.ExecuteActivity(workflow.WithActivityOptions(
2193
wfCtx, workflow.ActivityOptions{
2194
DisableEagerExecution: true,
2195
VersioningIntent: temporal.VersioningIntentCompatible,
2196
StartToCloseTimeout: 200 * time.Millisecond,
2197
RetryPolicy: &temporal.RetryPolicy{
2198
InitialInterval: 10 * time.Millisecond,
2199
BackoffCoefficient: 1,
2200
},
2201
}), "act", i*int(activityRuns)+j)
2202
futures = append(futures, f)
2203
}
2204
2205
for _, f := range futures {
2206
var activityVersion string
2207
err := f.Get(wfCtx, &activityVersion)
2208
s.NoError(err)
2209
res = append(res, activityVersion)
2210
// The output of a newer build ID should never be sent to a wf worker of an older build ID
2211
s.GreaterOrEqual(wfVersion, activityVersion)
2212
// TODO: uncomment this check once workflow.GetInfo(wfCtx).GetCurrentBuildID() returns correct value
2213
// based on last started task build ID, not last completed task build ID.
2214
// s.Assert().GreaterOrEqual(workflow.GetInfo(wfCtx).GetCurrentBuildID(), activityVersion)
2215
}
2216
}
2217
2218
return strings.Join(res, " "), nil
2219
}
2220
2221
// run all workers
2222
for i := 0; i <= 9; i++ {
2223
v := versions[i]
2224
w := worker.New(env.SdkClient(), tq, worker.Options{
2225
BuildID: v,
2226
UseBuildIDForVersioning: true,
2227
MaxConcurrentWorkflowTaskPollers: numPollers,
2228
MaxConcurrentActivityTaskPollers: 2,
2229
// Limit the number of concurrent activities so not all scheduled activities are immediately started.
2230
MaxConcurrentActivityExecutionSize: 2,
2231
})
2232
w.RegisterWorkflowWithOptions(
2233
func(ctx workflow.Context) (string, error) {
2234
return wf(ctx, v)
2235
},
2236
workflow.RegisterOptions{Name: "wf"})
2237
w.RegisterActivityWithOptions(
2238
func(runID int32) (string, error) {
2239
return act(v, runID)
2240
},
2241
activity.RegisterOptions{Name: "act"})
2242
s.NoError(w.Start())
2243
workers = append(workers, w)
2244
}
2245
2246
run, err := env.SdkClient().ExecuteWorkflow(s.Context(), sdkclient.StartWorkflowOptions{TaskQueue: tq}, "wf")
2247
s.NoError(err)
2248
2249
// Workflow should finish, otherwise it may mean we dropped some task without rescheduling them in the new build ID
2250
var out string
2251
s.NoError(run.Get(s.Context(), &out))
2252
s.validateWorkflowBuildIds(env, run.GetID(), run.GetRunID(), versions[9], true, versions[9], "", versions[:9])
2253
2254
activityPerVersion := make(map[string]int)
2255
for v := range strings.SplitSeq(out, " ") {
2256
activityPerVersion[v]++
2257
}
2258
2259
wh := env.SdkClient().GetWorkflowHistory(s.Context(), run.GetID(), run.GetRunID(), false, enumspb.HISTORY_EVENT_FILTER_TYPE_ALL_EVENT)
2260
redirectAppliedToActivityTask := false
2261
activityRetried := false
2262
sawUnorderedEvents := false
2263
var maxBuildId string
2264
var maxStartedTimestamp time.Time
2265
for wh.HasNext() {
2266
he, err := wh.Next()
2267
s.NoError(err)
2268
var taskStartedStamp *commonpb.WorkerVersionStamp
2269
var taskRedirectCounter int64
2270
var buildID string
2271
if activityStarted := he.GetActivityTaskStartedEventAttributes(); activityStarted != nil {
2272
taskStartedStamp = activityStarted.GetWorkerVersion()
2273
buildID = taskStartedStamp.GetBuildId()
2274
if buildID > maxBuildId {
2275
redirectAppliedToActivityTask = true
2276
}
2277
if activityStarted.Attempt > 1 {
2278
activityRetried = true
2279
}
2280
s.True(taskStartedStamp.GetUseVersioning())
2281
taskRedirectCounter = activityStarted.GetBuildIdRedirectCounter()
2282
activityPerVersion[buildID]--
2283
} else if wfStarted := he.GetWorkflowTaskStartedEventAttributes(); wfStarted != nil {
2284
taskStartedStamp = wfStarted.GetWorkerVersion()
2285
if taskStartedStamp != nil {
2286
// taskStartedStamp is nil for sticky queues
2287
s.True(taskStartedStamp.GetUseVersioning())
2288
buildID = taskStartedStamp.GetBuildId()
2289
taskRedirectCounter = wfStarted.GetBuildIdRedirectCounter()
2290
}
2291
}
2292
if he.EventTime.AsTime().Before(maxStartedTimestamp) {
2293
sawUnorderedEvents = true
2294
} else {
2295
maxStartedTimestamp = he.EventTime.AsTime()
2296
}
2297
if buildID > maxBuildId {
2298
maxBuildId = buildID
2299
}
2300
if taskStartedStamp != nil {
2301
// the last char of version is the index in the versions array which is the expected redirect counter for
2302
// a task started event
2303
expectedRedirectCounter, err := strconv.Atoi(buildID[len(buildID)-1:])
2304
s.NoError(err)
2305
s.Equal(expectedRedirectCounter, int(taskRedirectCounter))
2306
}
2307
}
2308
for v, c := range activityPerVersion {
2309
s.Equal(0, c, "activity count mismatch for build ID "+v)
2310
}
2311
// Following validations are more to make sure the test stays correct, rather than testing server's functionality
2312
s.True(activityRetried, "no activity retried")
2313
s.True(sawUnorderedEvents)
2314
s.True(redirectAppliedToActivityTask, "no redirect rule applied to an activity task, is this test broken?")
2315
2316
for _, w := range workers {
2317
w.Stop()
2318
}
2319
}
2320
2321
func (s *VersioningIntegSuite) TestDispatchActivityCompatible() {
2322
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) { s.dispatchActivityCompatible(env) })
2323
}
2324
2325
func (s *VersioningIntegSuite) dispatchActivityCompatible(env *testcore.TestEnv) {
2326
tq := testcore.RandomizeStr(s.T().Name())
2327
v1 := s.prefixed("v1")
2328
v11 := s.prefixed("v11")
2329
2330
started := make(chan struct{}, 2)
2331
2332
act1 := func() (string, error) { return "v1", nil }
2333
act11 := func() (string, error) { return "v1.1", nil }
2334
wf1 := func(ctx workflow.Context) (string, error) {
2335
started <- struct{}{}
2336
// wait for signal
2337
workflow.GetSignalChannel(ctx, "wait").Receive(ctx, nil)
2338
// run activity
2339
fut11 := workflow.ExecuteActivity(workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
2340
ScheduleToCloseTimeout: time.Minute,
2341
DisableEagerExecution: true,
2342
VersioningIntent: temporal.VersioningIntentCompatible,
2343
}), "act")
2344
var val11 string
2345
s.NoError(fut11.Get(ctx, &val11))
2346
return val11, nil
2347
}
2348
2349
s.addNewDefaultBuildID(env, tq, v1)
2350
s.waitForVersionSetPropagation(env, tq, v1)
2351
2352
w1 := worker.New(env.SdkClient(), tq, worker.Options{
2353
BuildID: v1,
2354
UseBuildIDForVersioning: true,
2355
MaxConcurrentWorkflowTaskPollers: numPollers,
2356
})
2357
w1.RegisterWorkflowWithOptions(wf1, workflow.RegisterOptions{Name: "wf"})
2358
w1.RegisterActivityWithOptions(act1, activity.RegisterOptions{Name: "act"})
2359
s.NoError(w1.Start())
2360
defer w1.Stop()
2361
2362
run, err := env.SdkClient().ExecuteWorkflow(s.Context(), sdkclient.StartWorkflowOptions{TaskQueue: tq}, "wf")
2363
s.NoError(err)
2364
// wait for it to start on v1
2365
env.WaitForChannel(started)
2366
2367
// now register v1.1 as compatible
2368
s.addCompatibleBuildID(env, tq, v11, v1, false)
2369
s.waitForVersionSetPropagation(env, tq, v11)
2370
// start worker for v1.1
2371
w11 := worker.New(env.SdkClient(), tq, worker.Options{
2372
BuildID: v11,
2373
UseBuildIDForVersioning: true,
2374
MaxConcurrentWorkflowTaskPollers: numPollers,
2375
})
2376
w11.RegisterWorkflowWithOptions(wf1, workflow.RegisterOptions{Name: "wf"})
2377
w11.RegisterActivityWithOptions(act11, activity.RegisterOptions{Name: "act"})
2378
s.NoError(w11.Start())
2379
defer w11.Stop()
2380
2381
// wait for w1 long polls to all time out
2382
time.Sleep(longPollTime) //nolint:forbidigo
2383
2384
// unblock the workflow
2385
s.NoError(env.SdkClient().SignalWorkflow(s.Context(), run.GetID(), run.GetRunID(), "wait", nil))
2386
2387
var out string
2388
s.NoError(run.Get(s.Context(), &out))
2389
s.Equal("v1.1", out)
2390
2391
s.validateWorkflowBuildIds(env, run.GetID(), run.GetRunID(), v11, false, v11, "", []string{v1})
2392
}
2393
2394
func (s *VersioningIntegSuite) TestDispatchActivityEager() {
2395
env := s.setupEnv()
2396
2397
tq := testcore.RandomizeStr(s.T().Name())
2398
v1 := s.prefixed("v1")
2399
2400
_, err := env.SdkClient().ExecuteWorkflow(s.Context(), sdkclient.StartWorkflowOptions{TaskQueue: tq}, "wf")
2401
s.NoError(err)
2402
2403
pollResponse, err := env.SdkClient().WorkflowService().PollWorkflowTaskQueue(s.Context(), &workflowservice.PollWorkflowTaskQueueRequest{
2404
Namespace: env.Namespace().String(),
2405
TaskQueue: &taskqueuepb.TaskQueue{Name: tq},
2406
Identity: "test",
2407
WorkerVersionCapabilities: &commonpb.WorkerVersionCapabilities{
2408
BuildId: v1,
2409
},
2410
})
2411
s.NoError(err)
2412
startToCloseTimeout := time.Minute
2413
2414
completionResponse, err := env.SdkClient().WorkflowService().RespondWorkflowTaskCompleted(s.Context(), &workflowservice.RespondWorkflowTaskCompletedRequest{
2415
Identity: "test",
2416
WorkerVersionStamp: &commonpb.WorkerVersionStamp{
2417
BuildId: v1,
2418
UseVersioning: true,
2419
},
2420
TaskToken: pollResponse.TaskToken,
2421
Commands: []*commandpb.Command{
2422
{
2423
CommandType: enumspb.COMMAND_TYPE_SCHEDULE_ACTIVITY_TASK,
2424
Attributes: &commandpb.Command_ScheduleActivityTaskCommandAttributes{
2425
ScheduleActivityTaskCommandAttributes: &commandpb.ScheduleActivityTaskCommandAttributes{
2426
ActivityId: "compatible",
2427
TaskQueue: &taskqueuepb.TaskQueue{
2428
Name: tq,
2429
},
2430
StartToCloseTimeout: durationpb.New(startToCloseTimeout),
2431
ActivityType: &commonpb.ActivityType{
2432
Name: "ignore",
2433
},
2434
RequestEagerExecution: true,
2435
UseWorkflowBuildId: true,
2436
},
2437
},
2438
},
2439
{
2440
CommandType: enumspb.COMMAND_TYPE_SCHEDULE_ACTIVITY_TASK,
2441
Attributes: &commandpb.Command_ScheduleActivityTaskCommandAttributes{
2442
ScheduleActivityTaskCommandAttributes: &commandpb.ScheduleActivityTaskCommandAttributes{
2443
ActivityId: "latest",
2444
TaskQueue: &taskqueuepb.TaskQueue{
2445
Name: tq,
2446
},
2447
StartToCloseTimeout: durationpb.New(startToCloseTimeout),
2448
ActivityType: &commonpb.ActivityType{
2449
Name: "ignore",
2450
},
2451
RequestEagerExecution: true,
2452
UseWorkflowBuildId: false,
2453
},
2454
},
2455
},
2456
},
2457
})
2458
s.NoError(err)
2459
s.Len(completionResponse.ActivityTasks, 1)
2460
s.Equal("compatible", completionResponse.ActivityTasks[0].ActivityId)
2461
}
2462
2463
func (s *VersioningIntegSuite) TestDispatchActivityCrossTQFails() {
2464
env := s.setupEnv(
2465
testcore.WithDynamicConfig(dynamicconfig.MatchingNumTaskqueueReadPartitions, 1),
2466
testcore.WithDynamicConfig(dynamicconfig.MatchingNumTaskqueueWritePartitions, 1),
2467
)
2468
2469
tq := testcore.RandomizeStr(s.T().Name())
2470
crosstq := testcore.RandomizeStr(s.T().Name())
2471
v1 := s.prefixed("v1")
2472
2473
act := func() (string, error) { return "v1", nil }
2474
wf := func(ctx workflow.Context) (string, error) {
2475
fut := workflow.ExecuteActivity(workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
2476
StartToCloseTimeout: 1 * time.Second,
2477
TaskQueue: crosstq,
2478
VersioningIntent: temporal.VersioningIntentCompatible,
2479
}), "act")
2480
var val string
2481
if err := fut.Get(ctx, &val); err != nil {
2482
return "", err
2483
}
2484
return val, nil
2485
}
2486
2487
s.addNewDefaultBuildID(env, tq, v1)
2488
s.addNewDefaultBuildID(env, crosstq, v1)
2489
s.waitForVersionSetPropagation(env, tq, v1)
2490
s.waitForVersionSetPropagation(env, crosstq, v1)
2491
2492
w1 := worker.New(env.SdkClient(), tq, worker.Options{
2493
BuildID: v1,
2494
UseBuildIDForVersioning: true,
2495
MaxConcurrentWorkflowTaskPollers: numPollers,
2496
})
2497
w1.RegisterWorkflowWithOptions(wf, workflow.RegisterOptions{Name: "wf"})
2498
s.NoError(w1.Start())
2499
defer w1.Stop()
2500
2501
w1cross := worker.New(env.SdkClient(), crosstq, worker.Options{
2502
BuildID: v1,
2503
UseBuildIDForVersioning: true,
2504
MaxConcurrentWorkflowTaskPollers: numPollers,
2505
})
2506
w1cross.RegisterActivityWithOptions(act, activity.RegisterOptions{Name: "act"})
2507
s.NoError(w1cross.Start())
2508
defer w1cross.Stop()
2509
2510
run, err := env.SdkClient().ExecuteWorkflow(s.Context(), sdkclient.StartWorkflowOptions{TaskQueue: tq}, "wf")
2511
s.NoError(err)
2512
2513
s.waitForWorkflowTaskFailedCause(env, run, enumspb.WORKFLOW_TASK_FAILED_CAUSE_BAD_SCHEDULE_ACTIVITY_ATTRIBUTES)
2514
s.NoError(env.SdkClient().TerminateWorkflow(s.Context(), run.GetID(), run.GetRunID(), "test complete", nil))
2515
}
2516
2517
func (s *VersioningIntegSuite) TestDispatchChildWorkflowOld() {
2518
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) { s.dispatchChildWorkflow(env, false, false) })
2519
}
2520
2521
func (s *VersioningIntegSuite) TestDispatchChildWorkflow() {
2522
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) { s.dispatchChildWorkflow(env, true, false) })
2523
}
2524
2525
func (s *VersioningIntegSuite) TestDispatchChildWorkflowCrossTq() {
2526
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) { s.dispatchChildWorkflow(env, true, true) })
2527
}
2528
2529
func (s *VersioningIntegSuite) dispatchChildWorkflow(
2530
env *testcore.TestEnv,
2531
newVersioning bool, crossTq bool,
2532
) {
2533
// This also implicitly tests that a workflow stays on a compatible version set if a new
2534
// incompatible set is registered, because wf2 just panics. It further tests that
2535
// stickiness on v1 is not broken by registering v2, because the channel send will panic on
2536
// replay after we close the channel.
2537
2538
tq := testcore.RandomizeStr(s.T().Name())
2539
childxTq := testcore.RandomizeStr(s.T().Name() + "child")
2540
v1 := s.prefixed("v1")
2541
v2 := s.prefixed("v2")
2542
inheritedBuildId := ""
2543
if newVersioning {
2544
inheritedBuildId = v1
2545
}
2546
2547
started := make(chan struct{}, 1)
2548
2549
validateChildBuild := func(cctx workflow.Context, expectedBuildId, expectedInheritedBuildId string) {
2550
exec := workflow.GetInfo(cctx).WorkflowExecution
2551
expectedStampBuildId := ""
2552
if !newVersioning && expectedBuildId == v1 {
2553
expectedStampBuildId = expectedBuildId
2554
}
2555
s.validateWorkflowBuildIds(env,
2556
exec.ID,
2557
exec.RunID,
2558
expectedBuildId,
2559
newVersioning,
2560
expectedStampBuildId,
2561
expectedInheritedBuildId,
2562
nil,
2563
)
2564
if newVersioning {
2565
s.validateWorkflowEventsVersionStamps(env, exec.ID, exec.RunID, []string{expectedBuildId}, expectedInheritedBuildId)
2566
}
2567
}
2568
2569
child1 := func(cctx workflow.Context) (string, error) {
2570
validateChildBuild(cctx, v1, inheritedBuildId)
2571
return "v1", nil
2572
}
2573
child2 := func(cctx workflow.Context) (string, error) {
2574
validateChildBuild(cctx, v2, "")
2575
return "v2", nil
2576
}
2577
child1xTq := func(cctx workflow.Context) (string, error) {
2578
validateChildBuild(cctx, v1, inheritedBuildId)
2579
return "v1xTq", nil
2580
}
2581
child2xTq := func(cctx workflow.Context) (string, error) {
2582
validateChildBuild(cctx, v2, "")
2583
return "v2xTq", nil
2584
}
2585
wf1 := func(ctx workflow.Context) (string, error) {
2586
started <- struct{}{}
2587
// wait for signal
2588
workflow.GetSignalChannel(ctx, "wait").Receive(ctx, nil)
2589
childTq := tq
2590
if crossTq {
2591
childTq = childxTq
2592
}
2593
// run two child workflows
2594
fut1 := workflow.ExecuteChildWorkflow(workflow.WithChildOptions(ctx, workflow.ChildWorkflowOptions{
2595
TaskQueue: childTq,
2596
VersioningIntent: temporal.VersioningIntentCompatible,
2597
}), "child")
2598
fut2 := workflow.ExecuteChildWorkflow(workflow.WithChildOptions(ctx, workflow.ChildWorkflowOptions{
2599
TaskQueue: childTq,
2600
VersioningIntent: temporal.VersioningIntentDefault, // this one should go to default
2601
}), "child")
2602
var val1, val2 string
2603
s.NoError(fut1.Get(ctx, &val1))
2604
s.NoError(fut2.Get(ctx, &val2))
2605
2606
return val1 + val2, nil
2607
}
2608
wf2 := func(ctx workflow.Context) (string, error) {
2609
panic("workflow should not run on v2")
2610
}
2611
2612
if newVersioning {
2613
rule := s.addAssignmentRule(env, tq, v1)
2614
s.waitForAssignmentRulePropagation(env, tq, rule)
2615
rule = s.addAssignmentRule(env, childxTq, v1)
2616
s.waitForAssignmentRulePropagation(env, childxTq, rule)
2617
} else {
2618
s.addNewDefaultBuildID(env, tq, v1)
2619
s.waitForVersionSetPropagation(env, tq, v1)
2620
}
2621
2622
w1 := worker.New(env.SdkClient(), tq, worker.Options{
2623
BuildID: v1,
2624
UseBuildIDForVersioning: true,
2625
MaxConcurrentWorkflowTaskPollers: numPollers,
2626
})
2627
w1.RegisterWorkflowWithOptions(wf1, workflow.RegisterOptions{Name: "wf"})
2628
w1.RegisterWorkflowWithOptions(child1, workflow.RegisterOptions{Name: "child"})
2629
s.NoError(w1.Start())
2630
defer w1.Stop()
2631
w1xTq := worker.New(env.SdkClient(), childxTq, worker.Options{
2632
BuildID: v1,
2633
UseBuildIDForVersioning: true,
2634
MaxConcurrentWorkflowTaskPollers: numPollers,
2635
})
2636
w1xTq.RegisterWorkflowWithOptions(child1xTq, workflow.RegisterOptions{Name: "child"})
2637
s.NoError(w1xTq.Start())
2638
defer w1xTq.Stop()
2639
2640
run, err := env.SdkClient().ExecuteWorkflow(s.Context(), sdkclient.StartWorkflowOptions{TaskQueue: tq}, "wf")
2641
s.NoError(err)
2642
// wait for it to start on v1
2643
env.WaitForChannel(started)
2644
close(started) // force panic if replayed
2645
2646
// now register v2 as default
2647
if newVersioning {
2648
rule := s.addAssignmentRule(env, tq, v2)
2649
s.waitForAssignmentRulePropagation(env, tq, rule)
2650
rule = s.addAssignmentRule(env, childxTq, v2)
2651
s.waitForAssignmentRulePropagation(env, childxTq, rule)
2652
} else {
2653
s.addNewDefaultBuildID(env, tq, v2)
2654
s.waitForVersionSetPropagation(env, tq, v2)
2655
}
2656
2657
// start worker for v2
2658
w2 := worker.New(env.SdkClient(), tq, worker.Options{
2659
BuildID: v2,
2660
UseBuildIDForVersioning: true,
2661
MaxConcurrentWorkflowTaskPollers: numPollers,
2662
})
2663
w2.RegisterWorkflowWithOptions(wf2, workflow.RegisterOptions{Name: "wf"})
2664
w2.RegisterWorkflowWithOptions(child2, workflow.RegisterOptions{Name: "child"})
2665
s.NoError(w2.Start())
2666
defer w2.Stop()
2667
w2xTq := worker.New(env.SdkClient(), childxTq, worker.Options{
2668
BuildID: v2,
2669
UseBuildIDForVersioning: true,
2670
MaxConcurrentWorkflowTaskPollers: numPollers,
2671
})
2672
w2xTq.RegisterWorkflowWithOptions(child2xTq, workflow.RegisterOptions{Name: "child"})
2673
s.NoError(w2xTq.Start())
2674
defer w2xTq.Stop()
2675
2676
// unblock the workflow
2677
s.NoError(env.SdkClient().SignalWorkflow(s.Context(), run.GetID(), run.GetRunID(), "wait", nil))
2678
2679
var out string
2680
s.NoError(run.Get(s.Context(), &out))
2681
if crossTq {
2682
s.Equal("v1xTqv2xTq", out)
2683
} else {
2684
s.Equal("v1v2", out)
2685
}
2686
2687
s.validateWorkflowBuildIds(env, run.GetID(), run.GetRunID(), v1, newVersioning, v1, "", nil)
2688
}
2689
2690
func (s *VersioningIntegSuite) TestDispatchChildWorkflowUpgradeOld() {
2691
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) { s.dispatchChildWorkflowUpgrade(env, false) })
2692
}
2693
2694
func (s *VersioningIntegSuite) TestDispatchChildWorkflowUpgrade() {
2695
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) { s.dispatchChildWorkflowUpgrade(env, true) })
2696
}
2697
2698
func (s *VersioningIntegSuite) dispatchChildWorkflowUpgrade(
2699
env *testcore.TestEnv,
2700
newVersioning bool,
2701
) {
2702
tq := testcore.RandomizeStr(s.T().Name())
2703
v1 := s.prefixed("v1")
2704
v11 := s.prefixed("v11")
2705
inheritedBuildId := ""
2706
if newVersioning {
2707
inheritedBuildId = v11
2708
}
2709
2710
started := make(chan struct{}, 2)
2711
2712
child1 := func(cctx workflow.Context) (string, error) {
2713
exec := workflow.GetInfo(cctx).WorkflowExecution
2714
expectedStampBuildId := v1
2715
if newVersioning {
2716
expectedStampBuildId = ""
2717
}
2718
s.validateWorkflowBuildIds(env, exec.ID, exec.RunID, v1, newVersioning, expectedStampBuildId, inheritedBuildId, nil)
2719
return "v1", nil
2720
}
2721
child11 := func(cctx workflow.Context) (string, error) {
2722
exec := workflow.GetInfo(cctx).WorkflowExecution
2723
expectedStampBuildId := v11
2724
if newVersioning {
2725
expectedStampBuildId = ""
2726
}
2727
s.validateWorkflowBuildIds(env, exec.ID, exec.RunID, v11, newVersioning, expectedStampBuildId, inheritedBuildId, nil)
2728
return "v1.1", nil
2729
}
2730
wf1 := func(ctx workflow.Context) (string, error) {
2731
started <- struct{}{}
2732
// wait for signal
2733
workflow.GetSignalChannel(ctx, "wait").Receive(ctx, nil)
2734
// run child
2735
fut11 := workflow.ExecuteChildWorkflow(workflow.WithChildOptions(ctx, workflow.ChildWorkflowOptions{}), "child")
2736
var val11 string
2737
s.NoError(fut11.Get(ctx, &val11))
2738
return val11, nil
2739
}
2740
2741
wf11 := func(ctx workflow.Context) (string, error) {
2742
started <- struct{}{}
2743
// wait for signal
2744
workflow.GetSignalChannel(ctx, "wait").Receive(ctx, nil)
2745
// run child
2746
fut11 := workflow.ExecuteChildWorkflow(workflow.WithChildOptions(ctx, workflow.ChildWorkflowOptions{}), "child")
2747
var val11 string
2748
s.NoError(fut11.Get(ctx, &val11))
2749
return val11, nil
2750
}
2751
2752
if newVersioning {
2753
rule := s.addAssignmentRule(env, tq, v1)
2754
s.waitForAssignmentRulePropagation(env, tq, rule)
2755
} else {
2756
s.addNewDefaultBuildID(env, tq, v1)
2757
s.waitForVersionSetPropagation(env, tq, v1)
2758
}
2759
2760
w1 := worker.New(env.SdkClient(), tq, worker.Options{
2761
BuildID: v1,
2762
UseBuildIDForVersioning: true,
2763
MaxConcurrentWorkflowTaskPollers: numPollers,
2764
})
2765
w1.RegisterWorkflowWithOptions(wf1, workflow.RegisterOptions{Name: "wf"})
2766
w1.RegisterWorkflowWithOptions(child1, workflow.RegisterOptions{Name: "child"})
2767
s.NoError(w1.Start())
2768
defer w1.Stop()
2769
2770
run, err := env.SdkClient().ExecuteWorkflow(s.Context(), sdkclient.StartWorkflowOptions{TaskQueue: tq}, "wf")
2771
s.NoError(err)
2772
// wait for it to start on v1
2773
env.WaitForChannel(started)
2774
2775
// now register v1.1 as compatible
2776
if newVersioning {
2777
rule := s.addRedirectRule(env, tq, v1, v11)
2778
s.waitForRedirectRulePropagation(env, tq, rule)
2779
} else {
2780
s.addCompatibleBuildID(env, tq, v11, v1, false)
2781
s.waitForVersionSetPropagation(env, tq, v11)
2782
}
2783
// start worker for v1.1
2784
w11 := worker.New(env.SdkClient(), tq, worker.Options{
2785
BuildID: v11,
2786
UseBuildIDForVersioning: true,
2787
MaxConcurrentWorkflowTaskPollers: numPollers,
2788
})
2789
w11.RegisterWorkflowWithOptions(wf11, workflow.RegisterOptions{Name: "wf"})
2790
w11.RegisterWorkflowWithOptions(child11, workflow.RegisterOptions{Name: "child"})
2791
s.NoError(w11.Start())
2792
defer w11.Stop()
2793
2794
// wait for w1 long polls to all time out
2795
time.Sleep(longPollTime) //nolint:forbidigo
2796
2797
// unblock the workflow
2798
s.NoError(env.SdkClient().SignalWorkflow(s.Context(), run.GetID(), run.GetRunID(), "wait", nil))
2799
2800
var out string
2801
s.NoError(run.Get(s.Context(), &out))
2802
s.Equal("v1.1", out)
2803
s.validateWorkflowBuildIds(env, run.GetID(), run.GetRunID(), v11, newVersioning, v11, "", []string{v1})
2804
}
2805
2806
func (s *VersioningIntegSuite) TestDispatchChildWorkflowCrossTQFails() {
2807
env := s.setupEnv(
2808
testcore.WithDynamicConfig(dynamicconfig.MatchingNumTaskqueueReadPartitions, 1),
2809
testcore.WithDynamicConfig(dynamicconfig.MatchingNumTaskqueueWritePartitions, 1),
2810
)
2811
2812
tq := testcore.RandomizeStr(s.T().Name())
2813
crosstq := testcore.RandomizeStr(s.T().Name())
2814
v1 := s.prefixed("v1")
2815
2816
child := func(ctx workflow.Context) (string, error) { return "v1", nil }
2817
wf := func(ctx workflow.Context) (string, error) {
2818
fut := workflow.ExecuteChildWorkflow(workflow.WithChildOptions(ctx, workflow.ChildWorkflowOptions{
2819
TaskQueue: crosstq,
2820
VersioningIntent: temporal.VersioningIntentCompatible,
2821
}), "child")
2822
var val string
2823
if err := fut.Get(ctx, &val); err != nil {
2824
return "", err
2825
}
2826
return val, nil
2827
}
2828
2829
s.addNewDefaultBuildID(env, tq, v1)
2830
s.addNewDefaultBuildID(env, crosstq, v1)
2831
s.waitForVersionSetPropagation(env, tq, v1)
2832
s.waitForVersionSetPropagation(env, crosstq, v1)
2833
2834
w1 := worker.New(env.SdkClient(), tq, worker.Options{
2835
BuildID: v1,
2836
UseBuildIDForVersioning: true,
2837
MaxConcurrentWorkflowTaskPollers: numPollers,
2838
})
2839
w1.RegisterWorkflowWithOptions(wf, workflow.RegisterOptions{Name: "wf"})
2840
s.NoError(w1.Start())
2841
defer w1.Stop()
2842
2843
w1cross := worker.New(env.SdkClient(), crosstq, worker.Options{
2844
BuildID: v1,
2845
UseBuildIDForVersioning: true,
2846
MaxConcurrentWorkflowTaskPollers: numPollers,
2847
})
2848
w1cross.RegisterWorkflowWithOptions(child, workflow.RegisterOptions{Name: "child"})
2849
s.NoError(w1cross.Start())
2850
defer w1cross.Stop()
2851
2852
run, err := env.SdkClient().ExecuteWorkflow(s.Context(), sdkclient.StartWorkflowOptions{TaskQueue: tq}, "wf")
2853
s.NoError(err)
2854
2855
s.waitForWorkflowTaskFailedCause(env, run, enumspb.WORKFLOW_TASK_FAILED_CAUSE_BAD_START_CHILD_EXECUTION_ATTRIBUTES)
2856
s.NoError(env.SdkClient().TerminateWorkflow(s.Context(), run.GetID(), run.GetRunID(), "test complete", nil))
2857
}
2858
2859
func (s *VersioningIntegSuite) waitForWorkflowTaskFailedCause(
2860
env *testcore.TestEnv,
2861
run sdkclient.WorkflowRun,
2862
cause enumspb.WorkflowTaskFailedCause,
2863
) {
2864
s.Await(func(s *VersioningIntegSuite) {
2865
events := env.GetHistory(env.Namespace().String(), &commonpb.WorkflowExecution{
2866
WorkflowId: run.GetID(),
2867
RunId: run.GetRunID(),
2868
})
2869
for _, event := range events {
2870
if attrs := event.GetWorkflowTaskFailedEventAttributes(); attrs.GetCause() == cause {
2871
return
2872
}
2873
}
2874
s.Failf("WorkflowTaskFailed event not found", "cause: %v", cause)
2875
}, 10*time.Second, 500*time.Millisecond)
2876
}
2877
2878
func (s *VersioningIntegSuite) TestDispatchQueryOld() {
2879
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) { s.dispatchQuery(env, false) })
2880
}
2881
2882
func (s *VersioningIntegSuite) TestDispatchQuery() {
2883
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) { s.dispatchQuery(env, true) })
2884
}
2885
2886
func (s *VersioningIntegSuite) dispatchQuery(
2887
env *testcore.TestEnv,
2888
newVersioning bool,
2889
) {
2890
tq := testcore.RandomizeStr(s.T().Name())
2891
v1 := s.prefixed("v1")
2892
v11 := s.prefixed("v11")
2893
v2 := s.prefixed("v2")
2894
2895
started := make(chan struct{}, 10)
2896
2897
wf1 := func(ctx workflow.Context) error {
2898
if err := workflow.SetQueryHandler(ctx, "query", func() (string, error) { return "v1", nil }); err != nil {
2899
return err
2900
}
2901
started <- struct{}{}
2902
workflow.GetSignalChannel(ctx, "wait").Receive(ctx, nil)
2903
return nil
2904
}
2905
wf11 := func(ctx workflow.Context) error {
2906
if err := workflow.SetQueryHandler(ctx, "query", func() (string, error) { return "v1.1", nil }); err != nil {
2907
return err
2908
}
2909
started <- struct{}{}
2910
workflow.GetSignalChannel(ctx, "wait").Receive(ctx, nil)
2911
return nil
2912
}
2913
wf2 := func(ctx workflow.Context) error {
2914
if err := workflow.SetQueryHandler(ctx, "query", func() (string, error) { return "v2", nil }); err != nil {
2915
return err
2916
}
2917
return nil
2918
}
2919
2920
if newVersioning {
2921
rule := s.addAssignmentRule(env, tq, v1)
2922
s.waitForAssignmentRulePropagation(env, tq, rule)
2923
} else {
2924
s.addNewDefaultBuildID(env, tq, v1)
2925
s.waitForVersionSetPropagation(env, tq, v1)
2926
}
2927
2928
w1 := worker.New(env.SdkClient(), tq, worker.Options{
2929
BuildID: v1,
2930
UseBuildIDForVersioning: true,
2931
MaxConcurrentWorkflowTaskPollers: numPollers,
2932
})
2933
w1.RegisterWorkflowWithOptions(wf1, workflow.RegisterOptions{Name: "wf"})
2934
s.NoError(w1.Start())
2935
defer w1.Stop()
2936
2937
run, err := env.SdkClient().ExecuteWorkflow(s.Context(), sdkclient.StartWorkflowOptions{TaskQueue: tq}, "wf")
2938
s.NoError(err)
2939
// wait for it to start on v1
2940
env.WaitForChannel(started)
2941
2942
if newVersioning {
2943
rule := s.addAssignmentRule(env, tq, v2)
2944
s.waitForAssignmentRulePropagation(env, tq, rule)
2945
rrule := s.addRedirectRule(env, tq, v1, v11)
2946
s.waitForRedirectRulePropagation(env, tq, rrule)
2947
} else {
2948
// now register v1.1 as compatible
2949
// now register v11 as newer compatible with v1 AND v2 as a new default
2950
s.addCompatibleBuildID(env, tq, v11, v1, false)
2951
s.addNewDefaultBuildID(env, tq, v2)
2952
s.waitForVersionSetPropagation(env, tq, v2)
2953
}
2954
// add another 100ms to make sure it got to sticky queues also
2955
time.Sleep(100 * time.Millisecond) //nolint:forbidigo
2956
2957
// start worker for v1.1 and v2
2958
w11 := worker.New(env.SdkClient(), tq, worker.Options{
2959
BuildID: v11,
2960
UseBuildIDForVersioning: true,
2961
MaxConcurrentWorkflowTaskPollers: numPollers,
2962
})
2963
w11.RegisterWorkflowWithOptions(wf11, workflow.RegisterOptions{Name: "wf"})
2964
s.NoError(w11.Start())
2965
defer w11.Stop()
2966
w2 := worker.New(env.SdkClient(), tq, worker.Options{
2967
BuildID: v2,
2968
UseBuildIDForVersioning: true,
2969
MaxConcurrentWorkflowTaskPollers: numPollers,
2970
})
2971
w2.RegisterWorkflowWithOptions(wf2, workflow.RegisterOptions{Name: "wf"})
2972
s.NoError(w2.Start())
2973
defer w2.Stop()
2974
2975
if !newVersioning {
2976
// wait for w1 long polls to all time out
2977
time.Sleep(longPollTime) //nolint:forbidigo
2978
}
2979
2980
// query
2981
val, err := env.SdkClient().QueryWorkflow(s.Context(), run.GetID(), run.GetRunID(), "query")
2982
s.NoError(err)
2983
var out string
2984
s.NoError(val.Get(&out))
2985
s.Equal("v1.1", out)
2986
2987
// let the workflow complete
2988
s.NoError(env.SdkClient().SignalWorkflow(s.Context(), run.GetID(), run.GetRunID(), "wait", nil))
2989
2990
// wait for completion
2991
s.NoError(run.Get(s.Context(), nil))
2992
2993
// query on closed workflow
2994
val, err = env.SdkClient().QueryWorkflow(s.Context(), run.GetID(), run.GetRunID(), "query")
2995
s.NoError(err)
2996
s.NoError(val.Get(&out))
2997
s.Equal("v1.1", out)
2998
2999
// start another wf on v2. should complete immediately.
3000
run2, err := env.SdkClient().ExecuteWorkflow(s.Context(), sdkclient.StartWorkflowOptions{TaskQueue: tq}, "wf")
3001
s.NoError(err)
3002
3003
// wait for completion
3004
s.NoError(run2.Get(s.Context(), nil))
3005
3006
// query on closed workflow
3007
val, err = env.SdkClient().QueryWorkflow(s.Context(), run2.GetID(), run2.GetRunID(), "query")
3008
s.NoError(err)
3009
s.NoError(val.Get(&out))
3010
s.Equal("v2", out)
3011
}
3012
3013
func (s *VersioningIntegSuite) TestDispatchContinueAsNewOld() {
3014
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) { s.dispatchContinueAsNew(env, false, false) })
3015
}
3016
3017
func (s *VersioningIntegSuite) TestDispatchContinueAsNew() {
3018
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) { s.dispatchContinueAsNew(env, true, false) })
3019
}
3020
3021
func (s *VersioningIntegSuite) TestDispatchContinueAsNewCrossTq() {
3022
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) { s.dispatchContinueAsNew(env, true, true) })
3023
}
3024
3025
func (s *VersioningIntegSuite) dispatchContinueAsNew(
3026
env *testcore.TestEnv,
3027
newVersioning bool, crossTq bool,
3028
) {
3029
tq := testcore.RandomizeStr(s.T().Name())
3030
canxTq := testcore.RandomizeStr(s.T().Name() + "CaN")
3031
v1 := s.prefixed("v1")
3032
v2 := s.prefixed("v2")
3033
inheritedBuildId := ""
3034
if newVersioning {
3035
inheritedBuildId = v1
3036
}
3037
3038
started1 := make(chan struct{}, 10)
3039
3040
wf1 := func(wctx workflow.Context, attempt int) (string, error) {
3041
switch attempt {
3042
case 0:
3043
s.Equal(tq, workflow.GetInfo(wctx).TaskQueueName)
3044
newCtx := workflow.WithWorkflowVersioningIntent(wctx, temporal.VersioningIntentCompatible)
3045
if crossTq {
3046
newCtx = workflow.WithWorkflowTaskQueue(newCtx, canxTq)
3047
}
3048
started1 <- struct{}{}
3049
return "", workflow.NewContinueAsNewError(newCtx, "wf", attempt+1)
3050
case 1:
3051
exec := workflow.GetInfo(wctx).WorkflowExecution
3052
expectedStampBuildId := v1
3053
if newVersioning {
3054
expectedStampBuildId = ""
3055
}
3056
s.validateWorkflowBuildIds(env, exec.ID, exec.RunID, v1, newVersioning, expectedStampBuildId, inheritedBuildId, nil)
3057
if newVersioning {
3058
s.validateWorkflowEventsVersionStamps(env, exec.ID, exec.RunID, []string{v1}, inheritedBuildId)
3059
}
3060
workflow.GetSignalChannel(wctx, "wait").Receive(wctx, nil)
3061
if crossTq {
3062
s.Equal(canxTq, workflow.GetInfo(wctx).TaskQueueName)
3063
} else {
3064
s.Equal(tq, workflow.GetInfo(wctx).TaskQueueName)
3065
}
3066
newCtx := workflow.WithWorkflowVersioningIntent(wctx, temporal.VersioningIntentDefault) // this one should go to default
3067
if crossTq {
3068
newCtx = workflow.WithWorkflowTaskQueue(newCtx, canxTq)
3069
}
3070
started1 <- struct{}{}
3071
return "", workflow.NewContinueAsNewError(newCtx, "wf", attempt+1)
3072
case 2:
3073
// return "done!", nil
3074
}
3075
panic("oops")
3076
}
3077
wf2 := func(wctx workflow.Context, attempt int) (string, error) {
3078
if attempt == 2 {
3079
exec := workflow.GetInfo(wctx).WorkflowExecution
3080
s.validateWorkflowBuildIds(env, exec.ID, exec.RunID, v2, newVersioning, "", "", nil)
3081
if crossTq {
3082
s.Equal(canxTq, workflow.GetInfo(wctx).TaskQueueName)
3083
} else {
3084
s.Equal(tq, workflow.GetInfo(wctx).TaskQueueName)
3085
}
3086
return "done!", nil
3087
}
3088
panic("oops")
3089
}
3090
3091
if newVersioning {
3092
rule := s.addAssignmentRule(env, tq, v1)
3093
s.waitForAssignmentRulePropagation(env, tq, rule)
3094
rule = s.addAssignmentRule(env, canxTq, v1)
3095
s.waitForAssignmentRulePropagation(env, canxTq, rule)
3096
} else {
3097
s.addNewDefaultBuildID(env, tq, v1)
3098
s.waitForVersionSetPropagation(env, tq, v1)
3099
}
3100
3101
w1 := worker.New(env.SdkClient(), tq, worker.Options{
3102
BuildID: v1,
3103
UseBuildIDForVersioning: true,
3104
MaxConcurrentWorkflowTaskPollers: numPollers,
3105
})
3106
w1.RegisterWorkflowWithOptions(wf1, workflow.RegisterOptions{Name: "wf"})
3107
s.NoError(w1.Start())
3108
defer w1.Stop()
3109
w1xTq := worker.New(env.SdkClient(), canxTq, worker.Options{
3110
BuildID: v1,
3111
UseBuildIDForVersioning: true,
3112
MaxConcurrentWorkflowTaskPollers: numPollers,
3113
})
3114
w1xTq.RegisterWorkflowWithOptions(wf1, workflow.RegisterOptions{Name: "wf"})
3115
s.NoError(w1xTq.Start())
3116
defer w1xTq.Stop()
3117
3118
run, err := env.SdkClient().ExecuteWorkflow(s.Context(), sdkclient.StartWorkflowOptions{TaskQueue: tq}, "wf")
3119
s.NoError(err)
3120
// wait for it to start on v1
3121
env.WaitForChannel(started1)
3122
3123
// now make v2 as a new default
3124
if newVersioning {
3125
rule := s.addAssignmentRule(env, tq, v2)
3126
s.waitForAssignmentRulePropagation(env, tq, rule)
3127
rule = s.addAssignmentRule(env, canxTq, v2)
3128
s.waitForAssignmentRulePropagation(env, canxTq, rule)
3129
} else {
3130
s.addNewDefaultBuildID(env, tq, v2)
3131
s.waitForVersionSetPropagation(env, tq, v2)
3132
}
3133
// add another 100ms to make sure it got to sticky queues also
3134
time.Sleep(100 * time.Millisecond) //nolint:forbidigo
3135
3136
s.validateWorkflowBuildIds(env, run.GetID(), run.GetRunID(), v1, newVersioning, v1, "", nil)
3137
3138
// start workers for v2
3139
w2 := worker.New(env.SdkClient(), tq, worker.Options{
3140
BuildID: v2,
3141
UseBuildIDForVersioning: true,
3142
MaxConcurrentWorkflowTaskPollers: numPollers,
3143
})
3144
w2.RegisterWorkflowWithOptions(wf2, workflow.RegisterOptions{Name: "wf"})
3145
s.NoError(w2.Start())
3146
defer w2.Stop()
3147
w2xTq := worker.New(env.SdkClient(), canxTq, worker.Options{
3148
BuildID: v2,
3149
UseBuildIDForVersioning: true,
3150
MaxConcurrentWorkflowTaskPollers: numPollers,
3151
})
3152
w2xTq.RegisterWorkflowWithOptions(wf2, workflow.RegisterOptions{Name: "wf"})
3153
s.NoError(w2xTq.Start())
3154
defer w2xTq.Stop()
3155
3156
if !newVersioning {
3157
// wait for w1 long polls to all time out
3158
time.Sleep(longPollTime) //nolint:forbidigo
3159
}
3160
3161
// unblock the workflow. it should get kicked off the sticky queue and replay on v1
3162
s.NoError(env.SdkClient().SignalWorkflow(s.Context(), run.GetID(), "", "wait", nil))
3163
// wait for it to start on v1
3164
env.WaitForChannel(started1)
3165
3166
var out string
3167
s.NoError(run.Get(s.Context(), &out))
3168
s.Equal("done!", out)
3169
s.validateWorkflowBuildIds(env, run.GetID(), "", v2, newVersioning, v2, "", nil)
3170
if newVersioning {
3171
s.validateWorkflowEventsVersionStamps(env, run.GetID(), "", []string{v2}, "")
3172
}
3173
}
3174
3175
func (s *VersioningIntegSuite) TestDispatchContinueAsNewUpgradeOld() {
3176
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) { s.dispatchContinueAsNewUpgrade(env, false) })
3177
}
3178
3179
func (s *VersioningIntegSuite) TestDispatchContinueAsNewUpgrade() {
3180
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) { s.dispatchContinueAsNewUpgrade(env, true) })
3181
}
3182
3183
func (s *VersioningIntegSuite) dispatchContinueAsNewUpgrade(
3184
env *testcore.TestEnv,
3185
newVersioning bool,
3186
) {
3187
tq := testcore.RandomizeStr(s.T().Name())
3188
v1 := s.prefixed("v1")
3189
v11 := s.prefixed("v11")
3190
v2 := s.prefixed("v2")
3191
3192
started1 := make(chan struct{}, 10)
3193
started11 := make(chan struct{}, 20)
3194
3195
wf1 := func(ctx workflow.Context, attempt int) (string, error) {
3196
started1 <- struct{}{}
3197
workflow.GetSignalChannel(ctx, "wait").Receive(ctx, nil)
3198
switch attempt {
3199
case 0:
3200
// return "", workflow.NewContinueAsNewError(ctx, "wf", attempt+1)
3201
case 1:
3202
// newCtx := workflow.WithWorkflowVersioningIntent(ctx, temporal.VersioningIntentDefault) // this one should go to default
3203
// return "", workflow.NewContinueAsNewError(newCtx, "wf", attempt+1)
3204
case 2:
3205
// return "done!", nil
3206
}
3207
panic("oops")
3208
}
3209
wf11 := func(ctx workflow.Context, attempt int) (string, error) {
3210
started11 <- struct{}{}
3211
workflow.GetSignalChannel(ctx, "wait").Receive(ctx, nil)
3212
switch attempt {
3213
case 0:
3214
return "", workflow.NewContinueAsNewError(ctx, "wf", attempt+1)
3215
case 1:
3216
newCtx := workflow.WithWorkflowVersioningIntent(ctx, temporal.VersioningIntentDefault) // this one should go to default
3217
return "", workflow.NewContinueAsNewError(newCtx, "wf", attempt+1)
3218
case 2:
3219
// return "done!", nil
3220
}
3221
panic("oops")
3222
}
3223
wf2 := func(ctx workflow.Context, attempt int) (string, error) {
3224
switch attempt {
3225
case 2:
3226
return "done!", nil
3227
}
3228
panic("oops")
3229
}
3230
3231
if newVersioning {
3232
rule := s.addAssignmentRule(env, tq, v1)
3233
s.waitForAssignmentRulePropagation(env, tq, rule)
3234
} else {
3235
s.addNewDefaultBuildID(env, tq, v1)
3236
s.waitForVersionSetPropagation(env, tq, v1)
3237
}
3238
3239
w1 := worker.New(env.SdkClient(), tq, worker.Options{
3240
BuildID: v1,
3241
UseBuildIDForVersioning: true,
3242
MaxConcurrentWorkflowTaskPollers: numPollers,
3243
})
3244
w1.RegisterWorkflowWithOptions(wf1, workflow.RegisterOptions{Name: "wf"})
3245
s.NoError(w1.Start())
3246
defer w1.Stop()
3247
3248
run, err := env.SdkClient().ExecuteWorkflow(s.Context(), sdkclient.StartWorkflowOptions{TaskQueue: tq}, "wf")
3249
s.NoError(err)
3250
// wait for it to start on v1
3251
env.WaitForChannel(started1)
3252
3253
// now register v11 as newer compatible with v1 AND v2 as a new default
3254
if newVersioning {
3255
rule := s.addRedirectRule(env, tq, v1, v11)
3256
s.waitForRedirectRulePropagation(env, tq, rule)
3257
rule2 := s.addAssignmentRule(env, tq, v2)
3258
s.waitForAssignmentRulePropagation(env, tq, rule2)
3259
} else {
3260
s.addCompatibleBuildID(env, tq, v11, v1, false)
3261
s.addNewDefaultBuildID(env, tq, v2)
3262
s.waitForVersionSetPropagation(env, tq, v2)
3263
}
3264
// add another 100ms to make sure it got to sticky queues also
3265
time.Sleep(100 * time.Millisecond) //nolint:forbidigo
3266
3267
// start workers for v11 and v2
3268
w11 := worker.New(env.SdkClient(), tq, worker.Options{
3269
BuildID: v11,
3270
UseBuildIDForVersioning: true,
3271
MaxConcurrentWorkflowTaskPollers: numPollers,
3272
})
3273
w11.RegisterWorkflowWithOptions(wf11, workflow.RegisterOptions{Name: "wf"})
3274
s.NoError(w11.Start())
3275
defer w11.Stop()
3276
3277
w2 := worker.New(env.SdkClient(), tq, worker.Options{
3278
BuildID: v2,
3279
UseBuildIDForVersioning: true,
3280
MaxConcurrentWorkflowTaskPollers: numPollers,
3281
})
3282
w2.RegisterWorkflowWithOptions(wf2, workflow.RegisterOptions{Name: "wf"})
3283
s.NoError(w2.Start())
3284
defer w2.Stop()
3285
3286
if !newVersioning {
3287
// wait for w1 long polls to all time out
3288
time.Sleep(longPollTime) //nolint:forbidigo
3289
}
3290
3291
// unblock the workflow. it should get kicked off the sticky queue and replay on v11
3292
s.NoError(env.SdkClient().SignalWorkflow(s.Context(), run.GetID(), "", "wait", nil))
3293
env.WaitForChannel(started11)
3294
3295
// then continue-as-new onto v11
3296
env.WaitForChannel(started11)
3297
3298
// initial run
3299
s.validateWorkflowBuildIds(env, run.GetID(), run.GetRunID(), v11, newVersioning, v11, "", []string{v1})
3300
3301
inheritedBuildId := ""
3302
if newVersioning {
3303
inheritedBuildId = v11
3304
}
3305
// first CaN
3306
s.validateWorkflowBuildIds(env, run.GetID(), "", v11, newVersioning, v11, inheritedBuildId, nil)
3307
3308
// unblock the second run. it should continue on v11 then continue-as-new onto v2, then
3309
// complete.
3310
s.NoError(env.SdkClient().SignalWorkflow(s.Context(), run.GetID(), "", "wait", nil))
3311
3312
// second CaN
3313
var out string
3314
s.NoError(run.Get(s.Context(), &out))
3315
s.Equal("done!", out)
3316
s.validateWorkflowBuildIds(env, run.GetID(), "", v2, newVersioning, v2, "", nil)
3317
}
3318
3319
func (s *VersioningIntegSuite) TestDispatchRetryOld() {
3320
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) { s.dispatchRetryOld(env) })
3321
}
3322
3323
func (s *VersioningIntegSuite) dispatchRetryOld(env *testcore.TestEnv) {
3324
tq := testcore.RandomizeStr(s.T().Name())
3325
v1 := s.prefixed("v1")
3326
v11 := s.prefixed("v11")
3327
v2 := s.prefixed("v2")
3328
3329
started1 := make(chan struct{}, 10)
3330
started11 := make(chan struct{}, 30)
3331
3332
wf1 := func(ctx workflow.Context) (string, error) {
3333
started1 <- struct{}{}
3334
workflow.GetSignalChannel(ctx, "wait").Receive(ctx, nil)
3335
switch workflow.GetInfo(ctx).Attempt {
3336
case 1:
3337
// return "", errors.New("try again")
3338
case 2:
3339
// return "", errors.New("try again")
3340
case 3:
3341
// return "done!", nil
3342
}
3343
panic("oops")
3344
}
3345
wf11 := func(ctx workflow.Context) (string, error) {
3346
started11 <- struct{}{}
3347
workflow.GetSignalChannel(ctx, "wait").Receive(ctx, nil)
3348
switch workflow.GetInfo(ctx).Attempt {
3349
case 1:
3350
return "", errors.New("try again") //nolint:err113
3351
case 2:
3352
return "", errors.New("try again") //nolint:err113
3353
case 3:
3354
return "done!", nil
3355
}
3356
panic("oops")
3357
}
3358
wf2 := func(ctx workflow.Context) (string, error) {
3359
panic("oops")
3360
}
3361
3362
s.addNewDefaultBuildID(env, tq, v1)
3363
s.waitForVersionSetPropagation(env, tq, v1)
3364
3365
w1 := worker.New(env.SdkClient(), tq, worker.Options{
3366
BuildID: v1,
3367
UseBuildIDForVersioning: true,
3368
MaxConcurrentWorkflowTaskPollers: numPollers,
3369
})
3370
w1.RegisterWorkflowWithOptions(wf1, workflow.RegisterOptions{Name: "wf"})
3371
s.NoError(w1.Start())
3372
defer w1.Stop()
3373
3374
run, err := env.SdkClient().ExecuteWorkflow(s.Context(), sdkclient.StartWorkflowOptions{
3375
TaskQueue: tq,
3376
RetryPolicy: &temporal.RetryPolicy{
3377
InitialInterval: 1000 * time.Millisecond,
3378
},
3379
}, "wf")
3380
s.NoError(err)
3381
// wait for it to start on v1
3382
env.WaitForChannel(started1)
3383
3384
// now register v11 as newer compatible with v1 AND v2 as a new default
3385
s.addCompatibleBuildID(env, tq, v11, v1, false)
3386
s.addNewDefaultBuildID(env, tq, v2)
3387
s.waitForVersionSetPropagation(env, tq, v2)
3388
// add another 100ms to make sure it got to sticky queues also
3389
time.Sleep(100 * time.Millisecond) //nolint:forbidigo
3390
3391
// start workers for v11 and v2
3392
w11 := worker.New(env.SdkClient(), tq, worker.Options{
3393
BuildID: v11,
3394
UseBuildIDForVersioning: true,
3395
MaxConcurrentWorkflowTaskPollers: numPollers,
3396
})
3397
w11.RegisterWorkflowWithOptions(wf11, workflow.RegisterOptions{Name: "wf"})
3398
s.NoError(w11.Start())
3399
defer w11.Stop()
3400
3401
w2 := worker.New(env.SdkClient(), tq, worker.Options{
3402
BuildID: v2,
3403
UseBuildIDForVersioning: true,
3404
MaxConcurrentWorkflowTaskPollers: numPollers,
3405
})
3406
w2.RegisterWorkflowWithOptions(wf2, workflow.RegisterOptions{Name: "wf"})
3407
s.NoError(w2.Start())
3408
defer w2.Stop()
3409
3410
// wait for w1 long polls to all time out
3411
time.Sleep(longPollTime) //nolint:forbidigo
3412
3413
// unblock the workflow. it should replay on v11 and then retry (on v11).
3414
s.NoError(env.SdkClient().SignalWorkflow(s.Context(), run.GetID(), "", "wait", nil))
3415
env.WaitForChannel(started11) // replay
3416
env.WaitForChannel(started11) // attempt 2
3417
3418
// now it's blocked in attempt 2. unblock it.
3419
s.NoError(env.SdkClient().SignalWorkflow(s.Context(), run.GetID(), "", "wait", nil))
3420
3421
// wait for attempt 3. unblock that and it should return.
3422
env.WaitForChannel(started11) // attempt 3
3423
s.NoError(env.SdkClient().SignalWorkflow(s.Context(), run.GetID(), "", "wait", nil))
3424
3425
var out string
3426
s.NoError(run.Get(s.Context(), &out))
3427
s.Equal("done!", out)
3428
s.validateWorkflowBuildIds(env, run.GetID(), run.GetRunID(), v11, false, v11, "", nil)
3429
}
3430
3431
func (s *VersioningIntegSuite) TestDispatchRetry() {
3432
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) { s.dispatchRetry(env) })
3433
}
3434
3435
func (s *VersioningIntegSuite) dispatchRetry(env *testcore.TestEnv) {
3436
tq := testcore.RandomizeStr(s.T().Name())
3437
v1 := s.prefixed("v1")
3438
v2 := s.prefixed("v2")
3439
3440
started1 := make(chan struct{}, 10)
3441
started2 := make(chan struct{}, 10)
3442
3443
wf1 := func(ctx workflow.Context) (string, error) {
3444
started1 <- struct{}{}
3445
workflow.GetSignalChannel(ctx, "wait").Receive(ctx, nil)
3446
if workflow.GetInfo(ctx).Attempt == 1 {
3447
return "", errors.New("try again") //nolint:err113
3448
}
3449
panic("oops")
3450
}
3451
wf2 := func(ctx workflow.Context) (string, error) {
3452
started2 <- struct{}{}
3453
workflow.GetSignalChannel(ctx, "wait").Receive(ctx, nil)
3454
switch workflow.GetInfo(ctx).Attempt {
3455
case 1:
3456
panic("oops")
3457
case 2:
3458
return "", errors.New("try again") //nolint:err113
3459
case 3:
3460
return "done!", nil
3461
}
3462
panic("oops")
3463
}
3464
3465
rule := s.addAssignmentRule(env, tq, v1)
3466
s.waitForAssignmentRulePropagation(env, tq, rule)
3467
3468
w1 := worker.New(env.SdkClient(), tq, worker.Options{
3469
BuildID: v1,
3470
UseBuildIDForVersioning: true,
3471
MaxConcurrentWorkflowTaskPollers: numPollers,
3472
})
3473
w1.RegisterWorkflowWithOptions(wf1, workflow.RegisterOptions{Name: "wf"})
3474
s.NoError(w1.Start())
3475
defer w1.Stop()
3476
3477
run, err := env.SdkClient().ExecuteWorkflow(s.Context(), sdkclient.StartWorkflowOptions{
3478
TaskQueue: tq,
3479
RetryPolicy: &temporal.RetryPolicy{
3480
InitialInterval: 1000 * time.Millisecond,
3481
},
3482
}, "wf")
3483
s.NoError(err)
3484
// wait for it to start on v1
3485
env.WaitForChannel(started1)
3486
3487
// now register v2 as a new default
3488
rule = s.addAssignmentRule(env, tq, v2)
3489
s.waitForAssignmentRulePropagation(env, tq, rule)
3490
// add another 100ms to make sure it got to sticky queues also
3491
time.Sleep(100 * time.Millisecond) //nolint:forbidigo
3492
3493
w2 := worker.New(env.SdkClient(), tq, worker.Options{
3494
BuildID: v2,
3495
UseBuildIDForVersioning: true,
3496
MaxConcurrentWorkflowTaskPollers: numPollers,
3497
})
3498
w2.RegisterWorkflowWithOptions(wf2, workflow.RegisterOptions{Name: "wf"})
3499
s.NoError(w2.Start())
3500
defer w2.Stop()
3501
3502
// unblock the workflow on v1
3503
s.NoError(env.SdkClient().SignalWorkflow(s.Context(), run.GetID(), "", "wait", nil))
3504
3505
env.WaitForChannel(started2) // attempt 2
3506
// now it's blocked in attempt 2. unblock it.
3507
s.NoError(env.SdkClient().SignalWorkflow(s.Context(), run.GetID(), "", "wait", nil))
3508
3509
// wait for attempt 3. unblock that and it should return.
3510
env.WaitForChannel(started2) // attempt 3
3511
s.NoError(env.SdkClient().SignalWorkflow(s.Context(), run.GetID(), "", "wait", nil))
3512
3513
var out string
3514
s.NoError(run.Get(s.Context(), &out))
3515
s.Equal("done!", out)
3516
s.validateWorkflowBuildIds(env, run.GetID(), run.GetRunID(), v2, true, v2, "", nil)
3517
}
3518
3519
func (s *VersioningIntegSuite) TestDispatchCronOld() {
3520
s.T().Skip("Skipping test since this tests old versioning behavior and also flakes")
3521
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) { s.dispatchCron(env, false) })
3522
}
3523
3524
func (s *VersioningIntegSuite) TestDispatchCron() {
3525
s.T().Skip("Skipping test since this tests old versioning behavior and also flakes")
3526
s.runTestWithMatchingBehavior(func(env *testcore.TestEnv, s *VersioningIntegSuite) { s.dispatchCron(env, true) })
3527
}
3528
3529
func (s *VersioningIntegSuite) dispatchCron(
3530
env *testcore.TestEnv,
3531
newVersioning bool,
3532
) {
3533
tq := testcore.RandomizeStr(s.T().Name())
3534
v1 := s.prefixed("v1")
3535
v11 := s.prefixed("v11")
3536
v2 := s.prefixed("v2")
3537
3538
var runIDs1 []string
3539
var runIDs2 []string
3540
3541
var runs1 atomic.Int32
3542
var runs11 atomic.Int32
3543
var runs2 atomic.Int32
3544
3545
wf1 := func(ctx workflow.Context) (string, error) {
3546
runs1.Add(1)
3547
runIDs1 = append(runIDs1, workflow.GetInfo(ctx).WorkflowExecution.RunID)
3548
return "ok", nil
3549
}
3550
wf11 := func(ctx workflow.Context) (string, error) {
3551
runs11.Add(1)
3552
return "ok", nil
3553
}
3554
wf2 := func(ctx workflow.Context) (string, error) {
3555
runs2.Add(1)
3556
runIDs2 = append(runIDs2, workflow.GetInfo(ctx).WorkflowExecution.RunID)
3557
return "ok", nil
3558
}
3559
3560
if newVersioning {
3561
rule := s.addAssignmentRule(env, tq, v1)
3562
s.waitForAssignmentRulePropagation(env, tq, rule)
3563
} else {
3564
s.addNewDefaultBuildID(env, tq, v1)
3565
s.waitForVersionSetPropagation(env, tq, v1)
3566
}
3567
3568
w1 := worker.New(env.SdkClient(), tq, worker.Options{
3569
BuildID: v1,
3570
UseBuildIDForVersioning: true,
3571
MaxConcurrentWorkflowTaskPollers: numPollers,
3572
})
3573
w1.RegisterWorkflowWithOptions(wf1, workflow.RegisterOptions{Name: "wf"})
3574
s.NoError(w1.Start())
3575
defer w1.Stop()
3576
3577
run, err := env.SdkClient().ExecuteWorkflow(s.Context(), sdkclient.StartWorkflowOptions{
3578
TaskQueue: tq,
3579
CronSchedule: "@every 1s",
3580
WorkflowExecutionTimeout: 7 * time.Second,
3581
}, "wf")
3582
s.NoError(err)
3583
3584
// give it >=3 runs on v1
3585
s.AwaitTrue(
3586
func() bool {
3587
return runs1.Load() >= int32(3)
3588
},
3589
6*time.Second,
3590
100*time.Millisecond,
3591
)
3592
3593
if newVersioning {
3594
rule := s.addAssignmentRule(env, tq, v2)
3595
s.waitForAssignmentRulePropagation(env, tq, rule)
3596
} else {
3597
// now register v11 as newer compatible with v1 AND v2 as a new default.
3598
// it will run on v2 instead of v11 because cron always starts on default.
3599
s.addCompatibleBuildID(env, tq, v11, v1, false)
3600
s.addNewDefaultBuildID(env, tq, v2)
3601
s.waitForVersionSetPropagation(env, tq, v2)
3602
}
3603
3604
// start workers for v11 and v2
3605
w11 := worker.New(env.SdkClient(), tq, worker.Options{
3606
BuildID: v11,
3607
UseBuildIDForVersioning: true,
3608
MaxConcurrentWorkflowTaskPollers: numPollers,
3609
})
3610
w11.RegisterWorkflowWithOptions(wf11, workflow.RegisterOptions{Name: "wf"})
3611
s.NoError(w11.Start())
3612
defer w11.Stop()
3613
3614
w2 := worker.New(env.SdkClient(), tq, worker.Options{
3615
BuildID: v2,
3616
UseBuildIDForVersioning: true,
3617
MaxConcurrentWorkflowTaskPollers: numPollers,
3618
})
3619
w2.RegisterWorkflowWithOptions(wf2, workflow.RegisterOptions{Name: "wf"})
3620
s.NoError(w2.Start())
3621
3622
// give it >=3 runs on v2
3623
s.AwaitTrue(
3624
func() bool {
3625
return runs2.Load() >= int32(3)
3626
},
3627
3500*time.Millisecond,
3628
100*time.Millisecond,
3629
)
3630
w2.Stop() // stop w2, because appending to runIDs2 while reading from it to validate causes a race
3631
s.Zero(runs11.Load())
3632
3633
for _, runid := range runIDs1 {
3634
s.validateWorkflowBuildIds(env, run.GetID(), runid, v1, newVersioning, v1, "", nil)
3635
}
3636
for _, runid := range runIDs2 {
3637
s.validateWorkflowBuildIds(env, run.GetID(), runid, v2, newVersioning, v2, "", nil)
3638
}
3639
}
3640
3641
func (s *VersioningIntegSuite) TestResetWorkflowAssignsToCorrectBuildId() {
3642
env := s.setupEnv()
3643
tq := testcore.RandomizeStr(s.T().Name())
3644
v1 := s.prefixed("v1")
3645
v2 := s.prefixed("v2")
3646
3647
act1 := func() (string, error) {
3648
return "act1 done!", nil
3649
}
3650
3651
wf := func(ctx workflow.Context) (string, error) {
3652
var ret string
3653
err := workflow.ExecuteActivity(
3654
workflow.WithActivityOptions(
3655
ctx, workflow.ActivityOptions{
3656
DisableEagerExecution: true,
3657
StartToCloseTimeout: 1 * time.Second,
3658
},
3659
),
3660
act1,
3661
).Get(ctx, &ret)
3662
s.NoError(err)
3663
s.Equal("act1 done!", ret)
3664
return "done!", nil
3665
}
3666
3667
rule := s.addAssignmentRule(env, tq, v1)
3668
s.waitForAssignmentRulePropagation(env, tq, rule)
3669
3670
w1 := worker.New(
3671
env.SdkClient(), tq, worker.Options{
3672
BuildID: v1,
3673
UseBuildIDForVersioning: true,
3674
MaxConcurrentWorkflowTaskPollers: numPollers,
3675
},
3676
)
3677
w1.RegisterWorkflow(wf)
3678
w1.RegisterActivity(act1)
3679
s.NoError(w1.Start())
3680
defer w1.Stop()
3681
w2 := worker.New(
3682
env.SdkClient(), tq, worker.Options{
3683
BuildID: v2,
3684
UseBuildIDForVersioning: true,
3685
MaxConcurrentWorkflowTaskPollers: numPollers,
3686
},
3687
)
3688
w2.RegisterWorkflow(wf)
3689
w2.RegisterActivity(act1)
3690
s.NoError(w2.Start())
3691
defer w2.Stop()
3692
3693
run, err := env.SdkClient().ExecuteWorkflow(s.Context(), sdkclient.StartWorkflowOptions{TaskQueue: tq}, wf)
3694
s.NoError(err)
3695
s.validateBuildIDAfterReset(env, run.GetID(), run.GetRunID(), false)
3696
}
3697
3698
func (s *VersioningIntegSuite) TestResetWorkflowAssignsToCorrectBuildId_CaN_Inherit() {
3699
env := s.setupEnv()
3700
s.resetWorkflowAssignsToCorrectBuildIDCan(env, true)
3701
}
3702
3703
func (s *VersioningIntegSuite) TestResetWorkflowAssignsToCorrectBuildId_CaN_NoInherit() {
3704
env := s.setupEnv()
3705
s.resetWorkflowAssignsToCorrectBuildIDCan(env, false)
3706
}
3707
3708
func (s *VersioningIntegSuite) resetWorkflowAssignsToCorrectBuildIDCan(
3709
env *testcore.TestEnv,
3710
inheritBuildID bool,
3711
) {
3712
tq := testcore.RandomizeStr(s.T().Name())
3713
v1 := s.prefixed("v1")
3714
v2 := s.prefixed("v2")
3715
3716
act1 := func() (string, error) {
3717
return "act1 done!", nil
3718
}
3719
3720
wf := func(ctx workflow.Context, attempt int) (string, error) {
3721
switch attempt {
3722
case 1:
3723
intent := temporal.VersioningIntentDefault
3724
if inheritBuildID {
3725
intent = temporal.VersioningIntentCompatible
3726
}
3727
newCtx := workflow.WithWorkflowVersioningIntent(ctx, intent)
3728
return "", workflow.NewContinueAsNewError(newCtx, "wf", 2)
3729
case 2:
3730
if workflow.GetInfo(ctx).Attempt == 1 {
3731
// failing first attempt of the CaN so we test inherit behavior across retry attempts
3732
return "", errors.New("try again") //nolint:err113
3733
}
3734
var ret string
3735
err := workflow.ExecuteActivity(
3736
workflow.WithActivityOptions(
3737
ctx, workflow.ActivityOptions{
3738
DisableEagerExecution: true,
3739
StartToCloseTimeout: 1 * time.Second,
3740
},
3741
),
3742
act1,
3743
).Get(ctx, &ret)
3744
s.NoError(err)
3745
s.Equal("act1 done!", ret)
3746
return "done!", nil
3747
default:
3748
panic("oops")
3749
}
3750
}
3751
3752
rule := s.addAssignmentRule(env, tq, v1)
3753
s.waitForAssignmentRulePropagation(env, tq, rule)
3754
3755
w1 := worker.New(
3756
env.SdkClient(), tq, worker.Options{
3757
BuildID: v1,
3758
UseBuildIDForVersioning: true,
3759
MaxConcurrentWorkflowTaskPollers: numPollers,
3760
},
3761
)
3762
w1.RegisterWorkflowWithOptions(wf, workflow.RegisterOptions{Name: "wf"})
3763
w1.RegisterActivity(act1)
3764
s.NoError(w1.Start())
3765
defer w1.Stop()
3766
w2 := worker.New(
3767
env.SdkClient(), tq, worker.Options{
3768
BuildID: v2,
3769
UseBuildIDForVersioning: true,
3770
MaxConcurrentWorkflowTaskPollers: numPollers,
3771
},
3772
)
3773
w2.RegisterWorkflowWithOptions(wf, workflow.RegisterOptions{Name: "wf"})
3774
w2.RegisterActivity(act1)
3775
s.NoError(w2.Start())
3776
defer w2.Stop()
3777
3778
run, err := env.SdkClient().ExecuteWorkflow(s.Context(), sdkclient.StartWorkflowOptions{
3779
TaskQueue: tq,
3780
RetryPolicy: &temporal.RetryPolicy{
3781
InitialInterval: 1000 * time.Millisecond,
3782
},
3783
}, wf, 1)
3784
s.NoError(err)
3785
s.validateBuildIDAfterReset(env, run.GetID(), "", inheritBuildID)
3786
}
3787
3788
func (s *VersioningIntegSuite) TestResetWorkflowAssignsToCorrectBuildId_ChildWF_Inherit() {
3789
env := s.setupEnv()
3790
s.resetWorkflowAssignsToCorrectBuildIDChildWf(env, true)
3791
}
3792
3793
func (s *VersioningIntegSuite) TestResetWorkflowAssignsToCorrectBuildId_ChildWF_NoInherit() {
3794
env := s.setupEnv()
3795
s.resetWorkflowAssignsToCorrectBuildIDChildWf(env, false)
3796
}
3797
3798
func (s *VersioningIntegSuite) resetWorkflowAssignsToCorrectBuildIDChildWf(
3799
env *testcore.TestEnv,
3800
inheritBuildID bool,
3801
) {
3802
tq := testcore.RandomizeStr(s.T().Name())
3803
v1 := s.prefixed("v1")
3804
v2 := s.prefixed("v2")
3805
3806
act1 := func() (string, error) {
3807
return "act1 done!", nil
3808
}
3809
3810
var childWfId string
3811
childStarted := make(chan struct{})
3812
3813
child := func(ctx workflow.Context) (string, error) {
3814
if workflow.GetInfo(ctx).Attempt == 1 {
3815
// failing first attempt of so we test inherit behavior across retry attempts
3816
return "", errors.New("try again") //nolint:err113
3817
}
3818
var ret string
3819
err := workflow.ExecuteActivity(
3820
workflow.WithActivityOptions(
3821
ctx, workflow.ActivityOptions{
3822
DisableEagerExecution: true,
3823
StartToCloseTimeout: 1 * time.Second,
3824
},
3825
),
3826
act1,
3827
).Get(ctx, &ret)
3828
s.NoError(err)
3829
s.Equal("act1 done!", ret)
3830
return "done!", nil
3831
}
3832
3833
wf := func(ctx workflow.Context) (string, error) {
3834
intent := temporal.VersioningIntentDefault
3835
if inheritBuildID {
3836
intent = temporal.VersioningIntentCompatible
3837
}
3838
fut := workflow.ExecuteChildWorkflow(workflow.WithChildOptions(ctx, workflow.ChildWorkflowOptions{
3839
VersioningIntent: intent,
3840
RetryPolicy: &temporal.RetryPolicy{
3841
InitialInterval: 1000 * time.Millisecond,
3842
},
3843
}), child)
3844
var val string
3845
var childWE workflow.Execution
3846
s.NoError(fut.GetChildWorkflowExecution().Get(ctx, &childWE))
3847
childWfId = childWE.ID
3848
close(childStarted)
3849
s.NoError(fut.Get(ctx, &val))
3850
s.Equal("done!", val)
3851
return "parent done!", nil
3852
}
3853
3854
rule := s.addAssignmentRule(env, tq, v1)
3855
s.waitForAssignmentRulePropagation(env, tq, rule)
3856
3857
w1 := worker.New(
3858
env.SdkClient(), tq, worker.Options{
3859
BuildID: v1,
3860
UseBuildIDForVersioning: true,
3861
MaxConcurrentWorkflowTaskPollers: numPollers,
3862
},
3863
)
3864
w1.RegisterWorkflow(wf)
3865
w1.RegisterWorkflow(child)
3866
w1.RegisterActivity(act1)
3867
s.NoError(w1.Start())
3868
defer w1.Stop()
3869
w2 := worker.New(
3870
env.SdkClient(), tq, worker.Options{
3871
BuildID: v2,
3872
UseBuildIDForVersioning: true,
3873
MaxConcurrentWorkflowTaskPollers: numPollers,
3874
},
3875
)
3876
w2.RegisterWorkflow(child)
3877
w2.RegisterActivity(act1)
3878
s.NoError(w2.Start())
3879
defer w2.Stop()
3880
3881
_, err := env.SdkClient().ExecuteWorkflow(s.Context(), sdkclient.StartWorkflowOptions{TaskQueue: tq}, wf)
3882
s.NoError(err)
3883
env.WaitForChannel(childStarted)
3884
s.validateBuildIDAfterReset(env, childWfId, "", inheritBuildID)
3885
}
3886
3887
// assumes given run has a single activity task.
3888
func (s *VersioningIntegSuite) validateBuildIDAfterReset(
3889
env *testcore.TestEnv,
3890
wfID, runID string, expectedInherit bool,
3891
) {
3892
v1 := s.prefixed("v1")
3893
v2 := s.prefixed("v2")
3894
3895
run := env.SdkClient().GetWorkflow(s.Context(), wfID, runID)
3896
3897
// let the original run finish
3898
var out string
3899
s.NoError(run.Get(s.Context(), &out))
3900
s.Equal("done!", out)
3901
3902
dw, err := env.SdkClient().DescribeWorkflowExecution(s.Context(), run.GetID(), run.GetRunID())
3903
s.NoError(err)
3904
tq := dw.GetWorkflowExecutionInfo().GetTaskQueue()
3905
inheritedBuildId := dw.GetWorkflowExecutionInfo().GetInheritedBuildId()
3906
s.Equal(expectedInherit, inheritedBuildId != "")
3907
s.validateWorkflowBuildIds(env, run.GetID(), run.GetRunID(), v1, true, v1, inheritedBuildId, nil)
3908
3909
// update rules with v2 as the default build
3910
rule := s.addAssignmentRule(env, tq, v2)
3911
s.waitForAssignmentRulePropagation(env, tq, rule)
3912
3913
// now reset the wf to first wf task
3914
wfr, err := env.SdkClient().ResetWorkflowExecution(s.Context(), &workflowservice.ResetWorkflowExecutionRequest{
3915
Namespace: env.Namespace().String(),
3916
WorkflowExecution: &commonpb.WorkflowExecution{
3917
WorkflowId: run.GetID(),
3918
RunId: run.GetRunID(),
3919
},
3920
WorkflowTaskFinishEventId: 3,
3921
})
3922
s.NoError(err)
3923
3924
// if a build ID is inherited, we should keep using that, otherwise should use the latest rules
3925
expectedBuildId := v2
3926
if inheritedBuildId != "" {
3927
expectedBuildId = inheritedBuildId
3928
}
3929
run2 := env.SdkClient().GetWorkflow(s.Context(), run.GetID(), wfr.GetRunId())
3930
s.NoError(run2.Get(s.Context(), &out))
3931
s.Equal("done!", out)
3932
s.validateWorkflowBuildIds(env, run2.GetID(), run2.GetRunID(), expectedBuildId, true, expectedBuildId, inheritedBuildId, nil)
3933
s.validateWorkflowEventsVersionStamps(env, run2.GetID(), run2.GetRunID(), []string{
3934
expectedBuildId,
3935
expectedBuildId,
3936
// expectedBuildId, skipped because it belongs to a sticky queue
3937
},
3938
inheritedBuildId)
3939
3940
// now reset the original wf to second wf task and make sure it remains in v1
3941
wfr, err = env.SdkClient().ResetWorkflowExecution(s.Context(), &workflowservice.ResetWorkflowExecutionRequest{
3942
Namespace: env.Namespace().String(),
3943
WorkflowExecution: &commonpb.WorkflowExecution{
3944
WorkflowId: run.GetID(),
3945
RunId: run.GetRunID(),
3946
},
3947
WorkflowTaskFinishEventId: 9,
3948
})
3949
s.NoError(err)
3950
3951
run3 := env.SdkClient().GetWorkflow(s.Context(), run.GetID(), wfr.GetRunId())
3952
s.NoError(run3.Get(s.Context(), &out))
3953
s.Equal("done!", out)
3954
s.validateWorkflowBuildIds(env, run3.GetID(), run3.GetRunID(), v1, true, v1, inheritedBuildId, nil)
3955
s.validateWorkflowEventsVersionStamps(env, run3.GetID(), run3.GetRunID(), []string{v1, v1, v1}, inheritedBuildId)
3956
}
3957
3958
func (s *VersioningIntegSuite) TestDescribeTaskQueueEnhanced_Versioned_ReachabilityCache() {
3959
env := s.setupEnv(s.reachabilityCacheTTLOptions()...)
3960
tq := testcore.RandomizeStr(s.T().Name())
3961
3962
// 1. Add assignment rule A and start workflow with build id A
3963
s.addAssignmentRule(env, tq, "A")
3964
started := make(chan struct{}, 10)
3965
wf := func(ctx workflow.Context) (string, error) {
3966
started <- struct{}{}
3967
workflow.GetSignalChannel(ctx, "wait").Receive(ctx, nil)
3968
if workflow.GetInfo(ctx).Attempt == 1 {
3969
return "", errors.New("try again") //nolint:err113
3970
}
3971
panic("oops")
3972
}
3973
wId := testcore.RandomizeStr("id")
3974
w := worker.New(env.SdkClient(), tq, worker.Options{
3975
UseBuildIDForVersioning: true,
3976
BuildID: "A",
3977
Identity: wId,
3978
})
3979
w.RegisterWorkflow(wf)
3980
s.NoError(w.Start())
3981
defer w.Stop()
3982
run, err := env.SdkClient().ExecuteWorkflow(s.Context(), sdkclient.StartWorkflowOptions{TaskQueue: tq}, wf)
3983
s.NoError(err)
3984
env.WaitForChannel(started)
3985
3986
// 2. Wait for visibility to show A as running with BuildId SearchAttribute 'assigned:A'
3987
s.Await(func(s *VersioningIntegSuite) {
3988
queryARunning := fmt.Sprintf("TaskQueue = '%s' AND BuildIds IN ('assigned:A') AND ExecutionStatus = \"Running\"", tq)
3989
resp, err := env.FrontendClient().CountWorkflowExecutions(s.Context(), &workflowservice.CountWorkflowExecutionsRequest{
3990
Namespace: env.Namespace().String(),
3991
Query: queryARunning,
3992
})
3993
s.NoError(err)
3994
s.Positive(resp.GetCount())
3995
}, 5*time.Second, 500*time.Millisecond)
3996
3997
// 3. Commit a different build id --> A should now only be reachable via visibility query
3998
s.commitBuildID(env, tq, "B", true, s.getVersioningRules(env, tq).GetConflictToken(), true)
3999
4000
// 4. Query reachability(A) --> reachable by visibility db, populating reachability open WF cache with A: true
4001
s.getBuildIDReachability(env, tq, &taskqueuepb.TaskQueueVersionSelection{BuildIds: []string{"A"}}, map[string]enumspb.BuildIdTaskReachability{
4002
"A": enumspb.BUILD_ID_TASK_REACHABILITY_REACHABLE, // reachable by visibility (db)
4003
})
4004
4005
// 5. Signal workflow with build id A so that it completes
4006
s.NoError(env.SdkClient().SignalWorkflow(s.Context(), run.GetID(), "", "wait", nil))
4007
4008
// 6. Query reachability(A) --> eventually shows closed_only by visibility db (after TTL passes and A is closed in visibility)
4009
s.AwaitTrue(func() bool {
4010
return s.checkBuildIDReachability(env, tq, &taskqueuepb.TaskQueueVersionSelection{BuildIds: []string{"A"}}, map[string]enumspb.BuildIdTaskReachability{
4011
"A": enumspb.BUILD_ID_TASK_REACHABILITY_CLOSED_WORKFLOWS_ONLY, // closed_only by visibility db (after TTL)
4012
})
4013
}, 5*time.Second, 50*time.Millisecond)
4014
}
4015
4016
func (s *VersioningIntegSuite) TestDescribeTaskQueueEnhanced_Versioned_BasicReachability() {
4017
env := s.setupEnv(s.reachabilityCacheTTLOptions()...)
4018
tq := testcore.RandomizeStr(s.T().Name())
4019
4020
s.getBuildIDReachability(env, tq, nil, map[string]enumspb.BuildIdTaskReachability{
4021
"": enumspb.BUILD_ID_TASK_REACHABILITY_REACHABLE, // reachable because unversioned is default
4022
})
4023
4024
s.addAssignmentRule(env, tq, "A")
4025
s.getBuildIDReachability(env, tq, &taskqueuepb.TaskQueueVersionSelection{BuildIds: []string{"", "A"}}, map[string]enumspb.BuildIdTaskReachability{
4026
"A": enumspb.BUILD_ID_TASK_REACHABILITY_REACHABLE, // reachable by default assignment rule
4027
"": enumspb.BUILD_ID_TASK_REACHABILITY_UNREACHABLE, // unreachable because no longer default
4028
})
4029
4030
// start workflow and worker with new default assignment rule "A", and wait for it to start
4031
started := make(chan struct{}, 10)
4032
wf := func(ctx workflow.Context) (string, error) {
4033
started <- struct{}{}
4034
workflow.GetSignalChannel(ctx, "wait").Receive(ctx, nil)
4035
if workflow.GetInfo(ctx).Attempt == 1 {
4036
return "", errors.New("try again") //nolint:err113
4037
}
4038
panic("oops")
4039
}
4040
wId := testcore.RandomizeStr("id")
4041
w := worker.New(env.SdkClient(), tq, worker.Options{
4042
UseBuildIDForVersioning: true,
4043
BuildID: "A",
4044
Identity: wId,
4045
})
4046
w.RegisterWorkflow(wf)
4047
s.NoError(w.Start())
4048
defer w.Stop()
4049
run, err := env.SdkClient().ExecuteWorkflow(s.Context(), sdkclient.StartWorkflowOptions{TaskQueue: tq}, wf)
4050
s.NoError(err)
4051
env.WaitForChannel(started)
4052
4053
// wait for visibility to show A as running with BuildId SearchAttribute 'assigned:A'
4054
s.Await(func(s *VersioningIntegSuite) {
4055
queryARunning := fmt.Sprintf("TaskQueue = '%s' AND BuildIds IN ('assigned:A') AND ExecutionStatus = \"Running\"", tq)
4056
resp, err := env.FrontendClient().CountWorkflowExecutions(s.Context(), &workflowservice.CountWorkflowExecutionsRequest{
4057
Namespace: env.Namespace().String(),
4058
Query: queryARunning,
4059
})
4060
s.NoError(err)
4061
s.Positive(resp.GetCount())
4062
}, 3*time.Second, 500*time.Millisecond)
4063
4064
// commit a different build ID --> A should now only be reachable via visibility query, B reachable as default
4065
s.commitBuildID(env, tq, "B", true, s.getVersioningRules(env, tq).GetConflictToken(), true)
4066
s.getBuildIDReachability(env, tq, nil, map[string]enumspb.BuildIdTaskReachability{
4067
"B": enumspb.BUILD_ID_TASK_REACHABILITY_REACHABLE, // reachable by default assignment rule
4068
})
4069
s.getBuildIDReachability(env, tq, &taskqueuepb.TaskQueueVersionSelection{BuildIds: []string{"A", "B"}}, map[string]enumspb.BuildIdTaskReachability{
4070
"A": enumspb.BUILD_ID_TASK_REACHABILITY_REACHABLE, // reachable by visibility
4071
"B": enumspb.BUILD_ID_TASK_REACHABILITY_REACHABLE, // reachable by default assignment rule
4072
})
4073
4074
// unblock the workflow on A so that it closes
4075
s.NoError(env.SdkClient().SignalWorkflow(s.Context(), run.GetID(), "", "wait", nil))
4076
4077
// Query reachability(A) --> eventually shows closed_only by visibility db (after TTL passes and A is closed in visibility)
4078
s.Await(func(s *VersioningIntegSuite) {
4079
s.True(s.checkBuildIDReachability(env, tq, &taskqueuepb.TaskQueueVersionSelection{BuildIds: []string{"A"}}, map[string]enumspb.BuildIdTaskReachability{
4080
"A": enumspb.BUILD_ID_TASK_REACHABILITY_CLOSED_WORKFLOWS_ONLY, // closed_only by visibility db (after TTL)
4081
"B": enumspb.BUILD_ID_TASK_REACHABILITY_REACHABLE, // reachable by default assignment rule
4082
}))
4083
}, 5*time.Second, 500*time.Millisecond)
4084
}
4085
4086
func (s *VersioningIntegSuite) TestDescribeTaskQueueEnhanced_Unversioned() {
4087
env := s.setupEnv()
4088
tq := testcore.RandomizeStr(s.T().Name())
4089
wf := func(ctx workflow.Context) (string, error) { return "ok", nil }
4090
4091
workerN := 3
4092
workerMap := make(map[string]worker.Worker)
4093
for range workerN {
4094
wId := testcore.RandomizeStr("id")
4095
w := worker.New(env.SdkClient(), tq, worker.Options{
4096
UseBuildIDForVersioning: false,
4097
Identity: wId,
4098
})
4099
w.RegisterWorkflow(wf)
4100
s.NoError(w.Start())
4101
defer w.Stop()
4102
workerMap[wId] = w
4103
}
4104
4105
s.Await(func(s *VersioningIntegSuite) {
4106
resp, err := env.FrontendClient().DescribeTaskQueue(s.Context(), &workflowservice.DescribeTaskQueueRequest{
4107
Namespace: env.Namespace().String(),
4108
TaskQueue: &taskqueuepb.TaskQueue{Name: tq, Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
4109
ApiMode: enumspb.DESCRIBE_TASK_QUEUE_MODE_ENHANCED,
4110
Versions: nil, // default version, in this case unversioned queue
4111
TaskQueueTypes: nil, // both types
4112
ReportPollers: true,
4113
ReportTaskReachability: true,
4114
ReportStats: false,
4115
})
4116
s.NoError(err)
4117
s.NotNil(resp)
4118
s.Len(resp.GetVersionsInfo(), 1, "should be 1 because only default/unversioned queue") //nolint:staticcheck
4119
versionInfo := resp.GetVersionsInfo()[""]
4120
s.Equal(enumspb.BUILD_ID_TASK_REACHABILITY_REACHABLE, versionInfo.GetTaskReachability())
4121
var pollersInfo []*taskqueuepb.PollerInfo
4122
for _, t := range versionInfo.GetTypesInfo() {
4123
pollersInfo = append(pollersInfo, t.GetPollers()...)
4124
}
4125
foundN := 0
4126
for wId := range workerMap {
4127
for _, pi := range pollersInfo {
4128
//nolint:staticcheck // SA1019: this test covers legacy build-ID versioning.
4129
s.False(pi.GetWorkerVersionCapabilities().GetUseVersioning())
4130
if pi.GetIdentity() == wId {
4131
foundN++
4132
break
4133
}
4134
}
4135
}
4136
4137
s.Equal(workerN, foundN)
4138
}, 3*time.Second, 500*time.Millisecond)
4139
}
4140
4141
func (s *VersioningIntegSuite) TestDescribeTaskQueueEnhanced_ReportFlags() {
4142
env := s.setupEnv()
4143
tq := testcore.RandomizeStr(s.T().Name())
4144
wf := func(ctx workflow.Context) (string, error) { return "ok", nil }
4145
4146
wId := testcore.RandomizeStr("id")
4147
w := worker.New(env.SdkClient(), tq, worker.Options{
4148
UseBuildIDForVersioning: false,
4149
Identity: wId,
4150
})
4151
w.RegisterWorkflow(wf)
4152
s.NoError(w.Start())
4153
defer w.Stop()
4154
4155
// wait for pollers to show up, verify both ReportPollers and ReportTaskReachability
4156
s.Await(func(s *VersioningIntegSuite) {
4157
resp, err := env.FrontendClient().DescribeTaskQueue(s.Context(), &workflowservice.DescribeTaskQueueRequest{
4158
Namespace: env.Namespace().String(),
4159
TaskQueue: &taskqueuepb.TaskQueue{Name: tq, Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
4160
ApiMode: enumspb.DESCRIBE_TASK_QUEUE_MODE_ENHANCED,
4161
Versions: nil, // default version, in this case unversioned queue
4162
TaskQueueTypes: nil, // both types
4163
ReportPollers: true,
4164
ReportTaskReachability: true,
4165
})
4166
s.NoError(err)
4167
s.NotNil(resp)
4168
s.Len(resp.GetVersionsInfo(), 1, "should be 1 because only default/unversioned queue") //nolint:staticcheck
4169
versionInfo := resp.GetVersionsInfo()[""]
4170
s.Equal(enumspb.BUILD_ID_TASK_REACHABILITY_REACHABLE, versionInfo.GetTaskReachability())
4171
var pollersInfo []*taskqueuepb.PollerInfo
4172
for _, t := range versionInfo.GetTypesInfo() {
4173
pollersInfo = append(pollersInfo, t.GetPollers()...)
4174
}
4175
foundPoller := false
4176
for _, pi := range pollersInfo {
4177
//nolint:staticcheck // SA1019: this test covers legacy build-ID versioning.
4178
s.False(pi.GetWorkerVersionCapabilities().GetUseVersioning())
4179
if pi.GetIdentity() == wId {
4180
foundPoller = true
4181
break
4182
}
4183
}
4184
s.True(foundPoller)
4185
}, 3*time.Second, 500*time.Millisecond)
4186
4187
// ask for reachability only
4188
resp, err := env.FrontendClient().DescribeTaskQueue(s.Context(), &workflowservice.DescribeTaskQueueRequest{
4189
Namespace: env.Namespace().String(),
4190
TaskQueue: &taskqueuepb.TaskQueue{Name: tq, Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
4191
ApiMode: enumspb.DESCRIBE_TASK_QUEUE_MODE_ENHANCED,
4192
Versions: nil, // default version, in this case unversioned queue
4193
TaskQueueTypes: nil, // both types
4194
ReportTaskReachability: true,
4195
})
4196
s.NoError(err)
4197
s.NotNil(resp)
4198
s.Len(resp.GetVersionsInfo(), 1, "should be 1 because only default/unversioned queue") //nolint:staticcheck
4199
versionInfo := resp.GetVersionsInfo()[""]
4200
s.Equal(enumspb.BUILD_ID_TASK_REACHABILITY_REACHABLE, versionInfo.GetTaskReachability())
4201
for _, t := range versionInfo.GetTypesInfo() {
4202
s.Empty(t.GetPollers(), "poller info should not be reported")
4203
}
4204
4205
// ask for pollers only
4206
resp, err = env.FrontendClient().DescribeTaskQueue(s.Context(), &workflowservice.DescribeTaskQueueRequest{
4207
Namespace: env.Namespace().String(),
4208
TaskQueue: &taskqueuepb.TaskQueue{Name: tq, Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
4209
ApiMode: enumspb.DESCRIBE_TASK_QUEUE_MODE_ENHANCED,
4210
Versions: nil, // default version, in this case unversioned queue
4211
TaskQueueTypes: nil, // both types
4212
ReportPollers: true,
4213
})
4214
s.NoError(err)
4215
s.NotNil(resp)
4216
s.Len(resp.GetVersionsInfo(), 1, "should be 1 because only default/unversioned queue") //nolint:staticcheck
4217
versionInfo = resp.GetVersionsInfo()[""]
4218
s.Equal(enumspb.BUILD_ID_TASK_REACHABILITY_UNSPECIFIED, versionInfo.GetTaskReachability())
4219
for _, t := range versionInfo.GetTypesInfo() {
4220
s.Len(t.GetPollers(), 1, "only one poller info should be reported")
4221
}
4222
}
4223
4224
func (s *VersioningIntegSuite) TestDescribeTaskQueueEnhanced_TooManyBuildIds() {
4225
env := s.setupEnv()
4226
tq := testcore.RandomizeStr(s.T().Name())
4227
4228
buildIDs := []string{"A", "B", "C", "D", "E"}
4229
resp, err := env.FrontendClient().DescribeTaskQueue(s.Context(), &workflowservice.DescribeTaskQueueRequest{
4230
Namespace: env.Namespace().String(),
4231
TaskQueue: &taskqueuepb.TaskQueue{Name: tq, Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
4232
ApiMode: enumspb.DESCRIBE_TASK_QUEUE_MODE_ENHANCED,
4233
Versions: &taskqueuepb.TaskQueueVersionSelection{BuildIds: buildIDs},
4234
TaskQueueTypes: nil, // both types
4235
ReportPollers: false,
4236
ReportTaskReachability: true,
4237
})
4238
s.NoError(err)
4239
s.NotNil(resp)
4240
4241
buildIDs = []string{"A", "B", "C", "D", "E", "F"}
4242
resp, err = env.FrontendClient().DescribeTaskQueue(s.Context(), &workflowservice.DescribeTaskQueueRequest{
4243
Namespace: env.Namespace().String(),
4244
TaskQueue: &taskqueuepb.TaskQueue{Name: tq, Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
4245
ApiMode: enumspb.DESCRIBE_TASK_QUEUE_MODE_ENHANCED,
4246
Versions: &taskqueuepb.TaskQueueVersionSelection{BuildIds: buildIDs},
4247
TaskQueueTypes: nil, // both types
4248
ReportPollers: false,
4249
ReportTaskReachability: true,
4250
})
4251
s.Error(err)
4252
s.Nil(resp)
4253
}
4254
4255
func (s *VersioningIntegSuite) TestDescribeTaskQueueLegacy_VersionSets() {
4256
// force one partition since DescribeTaskQueue only goes to the root
4257
env := s.setupEnv(
4258
testcore.WithDynamicConfig(dynamicconfig.MatchingNumTaskqueueReadPartitions, 1),
4259
testcore.WithDynamicConfig(dynamicconfig.MatchingNumTaskqueueWritePartitions, 1),
4260
)
4261
4262
tq := testcore.RandomizeStr(s.T().Name())
4263
v1 := s.prefixed("v1")
4264
v11 := s.prefixed("v11")
4265
v2 := s.prefixed("v2")
4266
4267
wf := func(ctx workflow.Context) (string, error) { return "ok", nil }
4268
4269
s.addNewDefaultBuildID(env, tq, v1)
4270
s.addCompatibleBuildID(env, tq, v11, v1, false)
4271
s.addNewDefaultBuildID(env, tq, v2)
4272
s.waitForVersionSetPropagation(env, tq, v2)
4273
4274
w1 := worker.New(env.SdkClient(), tq, worker.Options{
4275
BuildID: v1,
4276
UseBuildIDForVersioning: true,
4277
Identity: testcore.RandomizeStr("id"),
4278
})
4279
w1.RegisterWorkflow(wf)
4280
s.NoError(w1.Start())
4281
defer w1.Stop()
4282
4283
w11 := worker.New(env.SdkClient(), tq, worker.Options{
4284
BuildID: v11,
4285
UseBuildIDForVersioning: true,
4286
Identity: testcore.RandomizeStr("id"),
4287
})
4288
w11.RegisterWorkflow(wf)
4289
s.NoError(w11.Start())
4290
defer w11.Stop()
4291
4292
w2 := worker.New(env.SdkClient(), tq, worker.Options{
4293
BuildID: v2,
4294
UseBuildIDForVersioning: true,
4295
Identity: testcore.RandomizeStr("id"),
4296
})
4297
w2.RegisterWorkflow(wf)
4298
s.NoError(w2.Start())
4299
defer w2.Stop()
4300
4301
s.Await(func(s *VersioningIntegSuite) {
4302
resp, err := env.FrontendClient().DescribeTaskQueue(s.Context(), &workflowservice.DescribeTaskQueueRequest{
4303
Namespace: env.Namespace().String(),
4304
TaskQueue: &taskqueuepb.TaskQueue{Name: tq, Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
4305
TaskQueueType: enumspb.TASK_QUEUE_TYPE_WORKFLOW,
4306
})
4307
s.NoError(err)
4308
havePoller := func(v string) bool {
4309
for _, p := range resp.Pollers {
4310
if p.WorkerVersionCapabilities.UseVersioning && v == p.WorkerVersionCapabilities.BuildId {
4311
return true
4312
}
4313
}
4314
return false
4315
}
4316
// v1 polls get rejected because v11 is newer
4317
s.False(havePoller(v1))
4318
s.True(havePoller(v11))
4319
s.True(havePoller(v2))
4320
}, 3*time.Second, 500*time.Millisecond)
4321
}
4322
4323
func (s *VersioningIntegSuite) TestDescribeWorkflowExecution() {
4324
env := s.setupEnv(
4325
testcore.WithDynamicConfig(dynamicconfig.MatchingNumTaskqueueReadPartitions, 4),
4326
testcore.WithDynamicConfig(dynamicconfig.MatchingNumTaskqueueWritePartitions, 4),
4327
)
4328
4329
tq := testcore.RandomizeStr(s.T().Name())
4330
v1 := s.prefixed("v1")
4331
v11 := s.prefixed("v11")
4332
4333
started1 := make(chan struct{}, 10)
4334
started11 := make(chan struct{}, 10)
4335
4336
wf := func(ctx workflow.Context) (string, error) {
4337
started1 <- struct{}{}
4338
workflow.GetSignalChannel(ctx, "wait").Receive(ctx, nil)
4339
started11 <- struct{}{}
4340
workflow.GetSignalChannel(ctx, "wait").Receive(ctx, nil)
4341
return "ok", nil
4342
}
4343
4344
s.addNewDefaultBuildID(env, tq, v1)
4345
s.waitForVersionSetPropagation(env, tq, v1)
4346
4347
w1 := worker.New(env.SdkClient(), tq, worker.Options{
4348
BuildID: v1,
4349
UseBuildIDForVersioning: true,
4350
})
4351
w1.RegisterWorkflow(wf)
4352
s.NoError(w1.Start())
4353
defer w1.Stop()
4354
4355
run, err := env.SdkClient().ExecuteWorkflow(s.Context(), sdkclient.StartWorkflowOptions{TaskQueue: tq}, wf)
4356
s.NoError(err)
4357
// wait for it to start on v1
4358
env.WaitForChannel(started1)
4359
4360
// describe and check build ID
4361
s.Await(func(s *VersioningIntegSuite) {
4362
resp, err := env.SdkClient().DescribeWorkflowExecution(s.Context(), run.GetID(), "")
4363
s.NoError(err)
4364
//nolint:staticcheck // SA1019: this test covers legacy build-ID versioning.
4365
s.Equal(v1, resp.GetWorkflowExecutionInfo().GetMostRecentWorkerVersionStamp().GetBuildId())
4366
}, 5*time.Second, 500*time.Millisecond)
4367
4368
// now register v11 as newer compatible with v1
4369
s.addCompatibleBuildID(env, tq, v11, v1, false)
4370
s.waitForVersionSetPropagation(env, tq, v11)
4371
// add another 100ms to make sure it got to sticky queues also
4372
time.Sleep(100 * time.Millisecond) //nolint:forbidigo
4373
4374
// start worker for v11
4375
w11 := worker.New(env.SdkClient(), tq, worker.Options{
4376
BuildID: v11,
4377
UseBuildIDForVersioning: true,
4378
})
4379
w11.RegisterWorkflow(wf)
4380
s.NoError(w11.Start())
4381
defer w11.Stop()
4382
4383
// wait for w1 long polls to all time out
4384
time.Sleep(longPollTime) //nolint:forbidigo
4385
4386
// unblock the workflow. it should get kicked off the sticky queue and replay on v11
4387
s.NoError(env.SdkClient().SignalWorkflow(s.Context(), run.GetID(), "", "wait", nil))
4388
env.WaitForChannel(started11)
4389
4390
s.Await(func(s *VersioningIntegSuite) {
4391
resp, err := env.SdkClient().DescribeWorkflowExecution(s.Context(), run.GetID(), "")
4392
s.NoError(err)
4393
//nolint:staticcheck // SA1019: this test covers legacy build-ID versioning.
4394
s.Equal(v11, resp.GetWorkflowExecutionInfo().GetMostRecentWorkerVersionStamp().GetBuildId())
4395
}, 5*time.Second, 500*time.Millisecond)
4396
4397
// unblock. it should complete
4398
s.NoError(env.SdkClient().SignalWorkflow(s.Context(), run.GetID(), "", "wait", nil))
4399
var out string
4400
s.NoError(run.Get(s.Context(), &out))
4401
s.Equal("ok", out)
4402
}
4403
4404
// Add a per test prefix to avoid hitting the namespace limit of mapped task queue per build ID
4405
func (s *VersioningIntegSuite) prefixed(buildID string) string {
4406
return fmt.Sprintf("t%x:%s", 0xffff&farm.Hash32([]byte(s.T().Name())), buildID)
4407
}
4408
4409
// listVersioningRules lists rules and checks that the result is successful, returning the response.
4410
func (s *VersioningIntegSuite) getVersioningRules(
4411
env *testcore.TestEnv,
4412
tq string) *workflowservice.GetWorkerVersioningRulesResponse {
4413
res, err := env.FrontendClient().GetWorkerVersioningRules(s.Context(), &workflowservice.GetWorkerVersioningRulesRequest{
4414
Namespace: env.Namespace().String(),
4415
TaskQueue: tq,
4416
})
4417
s.NoError(err)
4418
s.NotNil(res)
4419
return res
4420
}
4421
4422
// insertAssignmentRule replaces a new versioning assignment rule to the task queue user data.
4423
// It checks the response and returns the conflict token.
4424
func (s *VersioningIntegSuite) insertAssignmentRule(
4425
env *testcore.TestEnv,
4426
tq, newBuildID string,
4427
idx int32, conflictToken []byte, expectSuccess bool) []byte {
4428
res, err := env.FrontendClient().UpdateWorkerVersioningRules(s.Context(), &workflowservice.UpdateWorkerVersioningRulesRequest{
4429
Namespace: env.Namespace().String(),
4430
TaskQueue: tq,
4431
ConflictToken: conflictToken,
4432
Operation: &workflowservice.UpdateWorkerVersioningRulesRequest_InsertAssignmentRule{
4433
InsertAssignmentRule: &workflowservice.UpdateWorkerVersioningRulesRequest_InsertBuildIdAssignmentRule{
4434
RuleIndex: idx,
4435
Rule: &taskqueuepb.BuildIdAssignmentRule{
4436
TargetBuildId: newBuildID,
4437
},
4438
},
4439
},
4440
})
4441
if expectSuccess {
4442
s.NoError(err)
4443
s.NotNil(res)
4444
s.Equal(newBuildID, res.GetAssignmentRules()[idx].GetRule().GetTargetBuildId())
4445
return res.GetConflictToken()
4446
} else {
4447
s.Error(err)
4448
s.Nil(res)
4449
return nil
4450
}
4451
}
4452
4453
// replaceAssignmentRule replaces a new versioning assignment rule to the task queue user data.
4454
// It checks the response and returns the conflict token.
4455
func (s *VersioningIntegSuite) replaceAssignmentRule(
4456
env *testcore.TestEnv,
4457
tq, newBuildID string,
4458
idx int32, conflictToken []byte, expectSuccess bool) []byte {
4459
res, err := env.FrontendClient().UpdateWorkerVersioningRules(s.Context(), &workflowservice.UpdateWorkerVersioningRulesRequest{
4460
Namespace: env.Namespace().String(),
4461
TaskQueue: tq,
4462
ConflictToken: conflictToken,
4463
Operation: &workflowservice.UpdateWorkerVersioningRulesRequest_ReplaceAssignmentRule{
4464
ReplaceAssignmentRule: &workflowservice.UpdateWorkerVersioningRulesRequest_ReplaceBuildIdAssignmentRule{
4465
RuleIndex: idx,
4466
Rule: &taskqueuepb.BuildIdAssignmentRule{
4467
TargetBuildId: newBuildID,
4468
},
4469
},
4470
},
4471
})
4472
if expectSuccess {
4473
s.NoError(err)
4474
s.NotNil(res)
4475
s.Equal(newBuildID, res.GetAssignmentRules()[idx].GetRule().GetTargetBuildId())
4476
return res.GetConflictToken()
4477
} else {
4478
s.Error(err)
4479
s.Nil(res)
4480
return nil
4481
}
4482
}
4483
4484
// deleteAssignmentRule deletes the versioning assignment rule at a given index.
4485
// It checks the response and returns the conflict token.
4486
func (s *VersioningIntegSuite) deleteAssignmentRule(
4487
env *testcore.TestEnv,
4488
tq string,
4489
idx int32, conflictToken []byte, expectSuccess bool) []byte {
4490
getResp, err := env.FrontendClient().GetWorkerVersioningRules(s.Context(), &workflowservice.GetWorkerVersioningRulesRequest{
4491
Namespace: env.Namespace().String(),
4492
TaskQueue: tq,
4493
})
4494
s.NoError(err)
4495
s.NotNil(getResp)
4496
4497
var prevRule *taskqueuepb.BuildIdAssignmentRule
4498
if expectSuccess {
4499
prevRule = getResp.GetAssignmentRules()[idx].GetRule()
4500
}
4501
4502
res, err := env.FrontendClient().UpdateWorkerVersioningRules(s.Context(), &workflowservice.UpdateWorkerVersioningRulesRequest{
4503
Namespace: env.Namespace().String(),
4504
TaskQueue: tq,
4505
ConflictToken: conflictToken,
4506
Operation: &workflowservice.UpdateWorkerVersioningRulesRequest_DeleteAssignmentRule{
4507
DeleteAssignmentRule: &workflowservice.UpdateWorkerVersioningRulesRequest_DeleteBuildIdAssignmentRule{
4508
RuleIndex: idx,
4509
},
4510
},
4511
})
4512
if expectSuccess {
4513
s.NoError(err)
4514
s.NotNil(res)
4515
found := false
4516
for _, r := range res.GetAssignmentRules() {
4517
if r.GetRule() == prevRule {
4518
found = true
4519
break
4520
}
4521
}
4522
s.False(found)
4523
return res.GetConflictToken()
4524
} else {
4525
s.Error(err)
4526
s.Nil(res)
4527
return nil
4528
}
4529
}
4530
4531
// insertRedirectRule replaces a new versioning redirect rule to the task queue user data.
4532
// It checks the response and returns the conflict token.
4533
func (s *VersioningIntegSuite) insertRedirectRule(
4534
env *testcore.TestEnv,
4535
tq, sourceBuildID, targetBuildID string,
4536
conflictToken []byte, expectSuccess bool) []byte {
4537
res, err := env.FrontendClient().UpdateWorkerVersioningRules(s.Context(), &workflowservice.UpdateWorkerVersioningRulesRequest{
4538
Namespace: env.Namespace().String(),
4539
TaskQueue: tq,
4540
ConflictToken: conflictToken,
4541
Operation: &workflowservice.UpdateWorkerVersioningRulesRequest_AddCompatibleRedirectRule{
4542
AddCompatibleRedirectRule: &workflowservice.UpdateWorkerVersioningRulesRequest_AddCompatibleBuildIdRedirectRule{
4543
Rule: &taskqueuepb.CompatibleBuildIdRedirectRule{
4544
SourceBuildId: sourceBuildID,
4545
TargetBuildId: targetBuildID,
4546
},
4547
},
4548
},
4549
})
4550
if expectSuccess {
4551
s.NoError(err)
4552
s.NotNil(res)
4553
found := false
4554
for _, r := range res.GetCompatibleRedirectRules() {
4555
if r.GetRule().GetSourceBuildId() == sourceBuildID && r.GetRule().GetTargetBuildId() == targetBuildID {
4556
found = true
4557
break
4558
}
4559
}
4560
s.True(found)
4561
return res.GetConflictToken()
4562
} else {
4563
s.Error(err)
4564
s.Nil(res)
4565
return nil
4566
}
4567
}
4568
4569
// replaceRedirectRule replaces a new versioning redirect rule to the task queue user data.
4570
// It checks the response and returns the conflict token.
4571
func (s *VersioningIntegSuite) replaceRedirectRule(
4572
env *testcore.TestEnv,
4573
tq, sourceBuildID, targetBuildID string,
4574
conflictToken []byte, expectSuccess bool) []byte {
4575
res, err := env.FrontendClient().UpdateWorkerVersioningRules(s.Context(), &workflowservice.UpdateWorkerVersioningRulesRequest{
4576
Namespace: env.Namespace().String(),
4577
TaskQueue: tq,
4578
ConflictToken: conflictToken,
4579
Operation: &workflowservice.UpdateWorkerVersioningRulesRequest_ReplaceCompatibleRedirectRule{
4580
ReplaceCompatibleRedirectRule: &workflowservice.UpdateWorkerVersioningRulesRequest_ReplaceCompatibleBuildIdRedirectRule{
4581
Rule: &taskqueuepb.CompatibleBuildIdRedirectRule{
4582
SourceBuildId: sourceBuildID,
4583
TargetBuildId: targetBuildID,
4584
},
4585
},
4586
},
4587
})
4588
if expectSuccess {
4589
s.NoError(err)
4590
s.NotNil(res)
4591
found := false
4592
for _, r := range res.GetCompatibleRedirectRules() {
4593
if r.GetRule().GetSourceBuildId() == sourceBuildID && r.GetRule().GetTargetBuildId() == targetBuildID {
4594
found = true
4595
break
4596
}
4597
}
4598
s.True(found)
4599
return res.GetConflictToken()
4600
} else {
4601
s.Error(err)
4602
s.Nil(res)
4603
return nil
4604
}
4605
}
4606
4607
// deleteRedirectRule deletes the versioning redirect rule at a given index.
4608
// It checks the response and returns the conflict token.
4609
func (s *VersioningIntegSuite) deleteRedirectRule(
4610
env *testcore.TestEnv,
4611
tq, sourceBuildID string,
4612
conflictToken []byte, expectSuccess bool) []byte {
4613
res, err := env.FrontendClient().UpdateWorkerVersioningRules(s.Context(), &workflowservice.UpdateWorkerVersioningRulesRequest{
4614
Namespace: env.Namespace().String(),
4615
TaskQueue: tq,
4616
ConflictToken: conflictToken,
4617
Operation: &workflowservice.UpdateWorkerVersioningRulesRequest_DeleteCompatibleRedirectRule{
4618
DeleteCompatibleRedirectRule: &workflowservice.UpdateWorkerVersioningRulesRequest_DeleteCompatibleBuildIdRedirectRule{
4619
SourceBuildId: sourceBuildID,
4620
},
4621
},
4622
})
4623
if expectSuccess {
4624
s.NoError(err)
4625
s.NotNil(res)
4626
found := false
4627
for _, r := range res.GetCompatibleRedirectRules() {
4628
if r.GetRule().GetSourceBuildId() == sourceBuildID {
4629
found = true
4630
break
4631
}
4632
}
4633
s.False(found)
4634
return res.GetConflictToken()
4635
} else {
4636
s.Error(err)
4637
s.Nil(res)
4638
return nil
4639
}
4640
}
4641
4642
// commitBuildID sends a CommitBuildId request for the given build ID
4643
// It checks the response and returns the conflict token.
4644
func (s *VersioningIntegSuite) commitBuildID(
4645
env *testcore.TestEnv,
4646
tq, targetBuildID string, force bool,
4647
conflictToken []byte, expectSuccess bool) []byte {
4648
res, err := env.FrontendClient().UpdateWorkerVersioningRules(s.Context(), &workflowservice.UpdateWorkerVersioningRulesRequest{
4649
Namespace: env.Namespace().String(),
4650
TaskQueue: tq,
4651
ConflictToken: conflictToken,
4652
Operation: &workflowservice.UpdateWorkerVersioningRulesRequest_CommitBuildId_{
4653
CommitBuildId: &workflowservice.UpdateWorkerVersioningRulesRequest_CommitBuildId{
4654
TargetBuildId: targetBuildID,
4655
Force: force,
4656
},
4657
},
4658
})
4659
if expectSuccess {
4660
s.NoError(err)
4661
s.NotNil(res)
4662
// 1. Adds a fully-ramped assignment rule for the target Build ID at the end of the list.
4663
endIdx := len(res.GetAssignmentRules()) - 1
4664
addedRule := res.GetAssignmentRules()[endIdx].GetRule()
4665
s.Equal(targetBuildID, addedRule.GetTargetBuildId())
4666
s.InEpsilon(float32(100), addedRule.GetPercentageRamp().GetRampPercentage(), 0.001)
4667
4668
foundOtherAssignmentRuleForTarget := false
4669
foundFullyRampedAssignmentRuleForOtherTarget := false
4670
for i, r := range res.GetAssignmentRules() {
4671
if r.GetRule().GetTargetBuildId() == targetBuildID && i != endIdx {
4672
foundOtherAssignmentRuleForTarget = true
4673
}
4674
if r.GetRule().GetPercentageRamp().GetRampPercentage() == 100 && r.GetRule().GetTargetBuildId() != targetBuildID {
4675
foundFullyRampedAssignmentRuleForOtherTarget = true
4676
}
4677
}
4678
// 2. Removes all previously added assignment rules to the given target Build ID (if any).
4679
s.False(foundOtherAssignmentRuleForTarget)
4680
// 3. Removes any fully-ramped assignment rule for other Build IDs.
4681
s.False(foundFullyRampedAssignmentRuleForOtherTarget)
4682
return res.GetConflictToken()
4683
} else {
4684
s.Error(err)
4685
s.Nil(res)
4686
return nil
4687
}
4688
}
4689
4690
func (s *VersioningIntegSuite) registerWorkflowAndPollVersionedTaskQueue(
4691
env *testcore.TestEnv,
4692
tq, buildID string, useVersioning bool,
4693
) {
4694
wf := func(ctx workflow.Context) (string, error) {
4695
return "done!", nil
4696
}
4697
4698
w1 := worker.New(env.SdkClient(), tq, worker.Options{
4699
BuildID: buildID,
4700
UseBuildIDForVersioning: useVersioning,
4701
MaxConcurrentWorkflowTaskPollers: numPollers,
4702
})
4703
w1.RegisterWorkflow(wf)
4704
s.NoError(w1.Start())
4705
defer w1.Stop()
4706
4707
// wait for it to start polling
4708
time.Sleep(200 * time.Millisecond) //nolint:forbidigo
4709
}
4710
4711
func (s *VersioningIntegSuite) getBuildIDReachability(
4712
env *testcore.TestEnv,
4713
taskQueue string,
4714
versions *taskqueuepb.TaskQueueVersionSelection,
4715
expectedReachability map[string]enumspb.BuildIdTaskReachability) {
4716
resp, err := env.FrontendClient().DescribeTaskQueue(s.Context(), &workflowservice.DescribeTaskQueueRequest{
4717
Namespace: env.Namespace().String(),
4718
TaskQueue: &taskqueuepb.TaskQueue{Name: taskQueue, Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
4719
ApiMode: enumspb.DESCRIBE_TASK_QUEUE_MODE_ENHANCED,
4720
Versions: versions,
4721
TaskQueueTypes: nil, // both types
4722
ReportPollers: false,
4723
ReportTaskReachability: true,
4724
})
4725
s.NoError(err)
4726
s.NotNil(resp)
4727
for buildID, vi := range resp.GetVersionsInfo() { //nolint:staticcheck
4728
expected, ok := expectedReachability[buildID]
4729
s.True(ok, "build id %s was not expected", buildID)
4730
s.Equal(expected, vi.GetTaskReachability(), "build id %s has unexpected reachability", buildID)
4731
}
4732
}
4733
4734
func (s *VersioningIntegSuite) checkBuildIDReachability(
4735
env *testcore.TestEnv,
4736
taskQueue string,
4737
versions *taskqueuepb.TaskQueueVersionSelection,
4738
expectedReachability map[string]enumspb.BuildIdTaskReachability) bool {
4739
resp, err := env.FrontendClient().DescribeTaskQueue(s.Context(), &workflowservice.DescribeTaskQueueRequest{
4740
Namespace: env.Namespace().String(),
4741
TaskQueue: &taskqueuepb.TaskQueue{Name: taskQueue, Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
4742
ApiMode: enumspb.DESCRIBE_TASK_QUEUE_MODE_ENHANCED,
4743
Versions: versions,
4744
TaskQueueTypes: nil, // both types
4745
ReportPollers: false,
4746
ReportTaskReachability: true,
4747
})
4748
if err != nil {
4749
return false
4750
}
4751
if resp == nil {
4752
return false
4753
}
4754
for buildID, vi := range resp.GetVersionsInfo() { //nolint:staticcheck
4755
expected, ok := expectedReachability[buildID]
4756
if !ok {
4757
return false // build id was not expected
4758
}
4759
if expected != vi.GetTaskReachability() {
4760
return false // build id has unexpected reachability
4761
}
4762
}
4763
return true
4764
}
4765
4766
// addNewDefaultBuildID updates build ID info on a task queue with a new build ID in a new default set.
4767
func (s *VersioningIntegSuite) addNewDefaultBuildID(
4768
env *testcore.TestEnv,
4769
tq, newBuildID string,
4770
) {
4771
res, err := env.FrontendClient().UpdateWorkerBuildIdCompatibility(s.Context(), &workflowservice.UpdateWorkerBuildIdCompatibilityRequest{
4772
Namespace: env.Namespace().String(),
4773
TaskQueue: tq,
4774
Operation: &workflowservice.UpdateWorkerBuildIdCompatibilityRequest_AddNewBuildIdInNewDefaultSet{
4775
AddNewBuildIdInNewDefaultSet: newBuildID,
4776
},
4777
})
4778
s.NoError(err)
4779
s.NotNil(res)
4780
}
4781
4782
func (s *VersioningIntegSuite) addAssignmentRule(
4783
env *testcore.TestEnv,
4784
tq, buildID string,
4785
) *taskqueuepb.BuildIdAssignmentRule {
4786
return s.addAssignmentRuleWithRamp(env, tq, buildID, 100)
4787
}
4788
4789
func (s *VersioningIntegSuite) addAssignmentRuleWithRamp(
4790
env *testcore.TestEnv,
4791
tq, buildID string, ramp float32,
4792
) *taskqueuepb.BuildIdAssignmentRule {
4793
rule := &taskqueuepb.BuildIdAssignmentRule{
4794
TargetBuildId: buildID,
4795
Ramp: &taskqueuepb.BuildIdAssignmentRule_PercentageRamp{
4796
PercentageRamp: &taskqueuepb.RampByPercentage{
4797
RampPercentage: ramp,
4798
},
4799
},
4800
}
4801
return s.doAddAssignmentRule(env, tq, rule)
4802
}
4803
4804
func (s *VersioningIntegSuite) doAddAssignmentRule(
4805
env *testcore.TestEnv,
4806
tq string, rule *taskqueuepb.BuildIdAssignmentRule,
4807
) *taskqueuepb.BuildIdAssignmentRule {
4808
cT := s.getVersioningRules(env, tq).GetConflictToken()
4809
res, err := env.FrontendClient().UpdateWorkerVersioningRules(s.Context(), &workflowservice.UpdateWorkerVersioningRulesRequest{
4810
Namespace: env.Namespace().String(),
4811
TaskQueue: tq,
4812
ConflictToken: cT,
4813
Operation: &workflowservice.UpdateWorkerVersioningRulesRequest_InsertAssignmentRule{
4814
InsertAssignmentRule: &workflowservice.UpdateWorkerVersioningRulesRequest_InsertBuildIdAssignmentRule{
4815
Rule: rule,
4816
},
4817
},
4818
})
4819
s.NoError(err)
4820
s.NotNil(res)
4821
return rule
4822
}
4823
4824
func (s *VersioningIntegSuite) addRedirectRule(
4825
env *testcore.TestEnv,
4826
tq, source string, target string,
4827
) *taskqueuepb.CompatibleBuildIdRedirectRule {
4828
cT := s.getVersioningRules(env, tq).GetConflictToken()
4829
rule := &taskqueuepb.CompatibleBuildIdRedirectRule{
4830
SourceBuildId: source,
4831
TargetBuildId: target,
4832
}
4833
res, err := env.FrontendClient().UpdateWorkerVersioningRules(s.Context(), &workflowservice.UpdateWorkerVersioningRulesRequest{
4834
Namespace: env.Namespace().String(),
4835
TaskQueue: tq,
4836
ConflictToken: cT,
4837
Operation: &workflowservice.UpdateWorkerVersioningRulesRequest_AddCompatibleRedirectRule{
4838
AddCompatibleRedirectRule: &workflowservice.UpdateWorkerVersioningRulesRequest_AddCompatibleBuildIdRedirectRule{
4839
Rule: rule,
4840
},
4841
},
4842
})
4843
s.NoError(err)
4844
s.NotNil(res)
4845
return rule
4846
}
4847
4848
func (s *VersioningIntegSuite) removeRedirectRule(
4849
env *testcore.TestEnv,
4850
tq, source string,
4851
) {
4852
cT := s.getVersioningRules(env, tq).GetConflictToken()
4853
res, err := env.FrontendClient().UpdateWorkerVersioningRules(s.Context(), &workflowservice.UpdateWorkerVersioningRulesRequest{
4854
Namespace: env.Namespace().String(),
4855
TaskQueue: tq,
4856
ConflictToken: cT,
4857
Operation: &workflowservice.UpdateWorkerVersioningRulesRequest_DeleteCompatibleRedirectRule{
4858
DeleteCompatibleRedirectRule: &workflowservice.UpdateWorkerVersioningRulesRequest_DeleteCompatibleBuildIdRedirectRule{
4859
SourceBuildId: source,
4860
},
4861
},
4862
})
4863
s.NoError(err)
4864
s.NotNil(res)
4865
}
4866
4867
// addCompatibleBuildID updates build ID info on a task queue with a new compatible build ID.
4868
func (s *VersioningIntegSuite) addCompatibleBuildID(
4869
env *testcore.TestEnv,
4870
tq, newBuildID, existing string, makeSetDefault bool,
4871
) {
4872
res, err := env.FrontendClient().UpdateWorkerBuildIdCompatibility(s.Context(), &workflowservice.UpdateWorkerBuildIdCompatibilityRequest{
4873
Namespace: env.Namespace().String(),
4874
TaskQueue: tq,
4875
Operation: &workflowservice.UpdateWorkerBuildIdCompatibilityRequest_AddNewCompatibleBuildId{
4876
AddNewCompatibleBuildId: &workflowservice.UpdateWorkerBuildIdCompatibilityRequest_AddNewCompatibleVersion{
4877
NewBuildId: newBuildID,
4878
ExistingCompatibleBuildId: existing,
4879
MakeSetDefault: makeSetDefault,
4880
},
4881
},
4882
})
4883
s.NoError(err)
4884
s.NotNil(res)
4885
}
4886
4887
// waitForVersionSetPropagation waits for all partitions of tq to mention newBuildID in their versioning data (in any position).
4888
func (s *VersioningIntegSuite) waitForVersionSetPropagation(
4889
env *testcore.TestEnv,
4890
taskQueue, newBuildID string,
4891
) {
4892
s.waitForPropagation(env, taskQueue, 0, func(vd *persistencespb.VersioningData) bool {
4893
for _, set := range vd.GetVersionSets() {
4894
for _, id := range set.BuildIds {
4895
if id.Id == newBuildID {
4896
return true
4897
}
4898
}
4899
}
4900
return false
4901
})
4902
}
4903
4904
// waitForAssignmentRulePropagation waits for all partitions of tq to have the given assignment rule in their versioning data
4905
func (s *VersioningIntegSuite) waitForAssignmentRulePropagation(
4906
env *testcore.TestEnv,
4907
taskQueue string, rule *taskqueuepb.BuildIdAssignmentRule,
4908
) {
4909
s.waitForPropagation(env, taskQueue, 0, func(vd *persistencespb.VersioningData) bool {
4910
for _, r := range vd.GetAssignmentRules() {
4911
if r.GetRule().Equal(rule) {
4912
return true
4913
}
4914
}
4915
return false
4916
})
4917
}
4918
4919
// waitForRedirectRulePropagation waits for all partitions of tq to have the given redirect rule in their versioning data
4920
func (s *VersioningIntegSuite) waitForRedirectRulePropagation(
4921
env *testcore.TestEnv,
4922
taskQueue string, rule *taskqueuepb.CompatibleBuildIdRedirectRule,
4923
) {
4924
s.waitForPropagation(env, taskQueue, 0, func(vd *persistencespb.VersioningData) bool {
4925
for _, r := range vd.GetRedirectRules() {
4926
if r.GetRule().Equal(rule) {
4927
return true
4928
}
4929
}
4930
return false
4931
})
4932
}
4933
4934
// waitForRedirectRulePropagationUpToPartition waits for partitions of tq, up to a certain partition ID, to have the
4935
// given redirect rule in their versioning data
4936
func (s *VersioningIntegSuite) waitForRedirectRulePropagationUpToPartition(
4937
env *testcore.TestEnv,
4938
taskQueue string,
4939
rule *taskqueuepb.CompatibleBuildIdRedirectRule,
4940
upToPartition int,
4941
) {
4942
s.waitForPropagation(env, taskQueue, upToPartition+1, func(vd *persistencespb.VersioningData) bool {
4943
for _, r := range vd.GetRedirectRules() {
4944
if r.GetRule().Equal(rule) {
4945
return true
4946
}
4947
}
4948
return false
4949
})
4950
}
4951
4952
// waitForPropagation waits for all partitions of tq to mention newBuildID in their versioning data (in any position).
4953
// Pass 0 for partitionCount to make it load from MatchingNumTaskqueueReadPartitions config.
4954
func (s *VersioningIntegSuite) waitForPropagation(
4955
env *testcore.TestEnv,
4956
taskQueue string,
4957
partitionCount int,
4958
condition func(data *persistencespb.VersioningData) bool,
4959
) {
4960
if partitionCount <= 0 {
4961
v := env.GetTestCluster().Host().DcClient().GetValue(dynamicconfig.MatchingNumTaskqueueReadPartitions.Key())
4962
s.NotEmpty(v, "versioning tests require setting explicit number of partitions")
4963
count, ok := v[0].Value.(int)
4964
s.True(ok, "partition count is not an int")
4965
partitionCount = count
4966
}
4967
4968
type partAndType struct {
4969
part int
4970
tp enumspb.TaskQueueType
4971
}
4972
remaining := make(map[partAndType]struct{})
4973
for i := 0; i < partitionCount; i++ {
4974
remaining[partAndType{i, enumspb.TASK_QUEUE_TYPE_ACTIVITY}] = struct{}{}
4975
remaining[partAndType{i, enumspb.TASK_QUEUE_TYPE_WORKFLOW}] = struct{}{}
4976
}
4977
s.Await(func(s *VersioningIntegSuite) {
4978
for pt := range remaining {
4979
f, err := tqid.NewTaskQueueFamily(env.NamespaceID().String(), taskQueue)
4980
s.NoError(err)
4981
partition := f.TaskQueue(pt.tp).NormalPartition(pt.part)
4982
// Use lower-level GetTaskQueueUserData instead of GetWorkerBuildIdCompatibility
4983
// here so that we can target activity queues.
4984
res, err := env.GetTestCluster().Host().MatchingClient().GetTaskQueueUserData(
4985
s.Context(),
4986
&matchingservice.GetTaskQueueUserDataRequest{
4987
NamespaceId: env.NamespaceID().String(),
4988
TaskQueue: partition.RpcName(),
4989
TaskQueueType: partition.TaskType(),
4990
})
4991
s.NoError(err)
4992
if condition(res.GetUserData().GetData().GetVersioningData()) {
4993
delete(remaining, pt)
4994
}
4995
}
4996
s.Empty(remaining)
4997
}, 10*time.Second, 500*time.Millisecond)
4998
}
4999
5000
func (s *VersioningIntegSuite) unloadTaskQueue(
5001
env *testcore.TestEnv,
5002
tq string,
5003
) {
5004
_, err := env.GetTestCluster().MatchingClient().ForceUnloadTaskQueuePartition(s.Context(), &matchingservice.ForceUnloadTaskQueuePartitionRequest{
5005
NamespaceId: env.NamespaceID().String(),
5006
TaskQueuePartition: &taskqueuespb.TaskQueuePartition{
5007
TaskQueue: tq,
5008
TaskQueueType: enumspb.TASK_QUEUE_TYPE_WORKFLOW,
5009
},
5010
})
5011
s.NoError(err)
5012
}
5013
5014
func (s *VersioningIntegSuite) getStickyQueueName(
5015
env *testcore.TestEnv,
5016
id string,
5017
) string {
5018
ms, err := env.AdminClient().DescribeMutableState(s.Context(), &adminservice.DescribeMutableStateRequest{
5019
Namespace: env.Namespace().String(),
5020
Execution: &commonpb.WorkflowExecution{WorkflowId: id},
5021
Archetype: chasm.WorkflowArchetype,
5022
})
5023
s.NoError(err)
5024
return ms.DatabaseMutableState.ExecutionInfo.StickyTaskQueue
5025
}
5026
5027
func getCurrentDefault(res *workflowservice.GetWorkerBuildIdCompatibilityResponse) string {
5028
if res == nil {
5029
return ""
5030
}
5031
curMajorSet := res.GetMajorVersionSets()[len(res.GetMajorVersionSets())-1]
5032
return curMajorSet.GetBuildIds()[len(curMajorSet.GetBuildIds())-1]
5033
}
5034
5035
// Periodically checks a WF and unblocks when it is assigned to the given build ID
5036
func (s *VersioningIntegSuite) waitForWorkflowBuildID(
5037
env *testcore.TestEnv,
5038
wfID string,
5039
runID string,
5040
buildID string,
5041
) {
5042
s.AwaitTrue(
5043
func() bool {
5044
dw, err := env.SdkClient().DescribeWorkflowExecution(s.Context(), wfID, runID)
5045
if err != nil {
5046
return false
5047
}
5048
return dw.GetWorkflowExecutionInfo().GetAssignedBuildId() == buildID //nolint:staticcheck
5049
},
5050
10*time.Second,
5051
100*time.Millisecond,
5052
)
5053
}
5054
5055
func (s *VersioningIntegSuite) validateWorkflowBuildIds(
5056
env *testcore.TestEnv,
5057
wfID string,
5058
runID string,
5059
expectedBuildId string,
5060
newVersioning bool,
5061
expectedStampBuildId string,
5062
expectedInheritedBuildId string,
5063
extraSearchAttrBuildIds []string,
5064
) {
5065
dw, err := env.SdkClient().DescribeWorkflowExecution(s.Context(), wfID, runID)
5066
s.NoError(err)
5067
saPayload := dw.GetWorkflowExecutionInfo().GetSearchAttributes().GetIndexedFields()["BuildIds"]
5068
searchAttrAny, err := sadefs.DecodeValue(saPayload, enumspb.INDEXED_VALUE_TYPE_KEYWORD_LIST, false)
5069
var searchAttr []string
5070
if searchAttrAny != nil {
5071
searchAttr = searchAttrAny.([]string)
5072
}
5073
s.NoError(err)
5074
if expectedBuildId == "" {
5075
if expectedStampBuildId != "" {
5076
s.NotNil(dw.GetWorkflowExecutionInfo().GetMostRecentWorkerVersionStamp().GetBuildId())
5077
s.False(dw.GetWorkflowExecutionInfo().GetMostRecentWorkerVersionStamp().GetUseVersioning())
5078
s.Len(searchAttr, 2+len(extraSearchAttrBuildIds))
5079
s.Equal(worker_versioning.UnversionedSearchAttribute, searchAttr[0])
5080
s.True(strings.HasPrefix(searchAttr[1], worker_versioning.UnversionedSearchAttribute))
5081
} else {
5082
s.Nil(dw.GetWorkflowExecutionInfo().GetMostRecentWorkerVersionStamp())
5083
s.Empty(searchAttr)
5084
}
5085
} else {
5086
if expectedStampBuildId != "" {
5087
s.True(dw.GetWorkflowExecutionInfo().GetMostRecentWorkerVersionStamp().GetUseVersioning())
5088
s.Equal(expectedStampBuildId, dw.GetWorkflowExecutionInfo().GetMostRecentWorkerVersionStamp().GetBuildId())
5089
} else {
5090
s.Nil(dw.GetWorkflowExecutionInfo().GetMostRecentWorkerVersionStamp())
5091
}
5092
if newVersioning {
5093
s.Equal(expectedBuildId, dw.GetWorkflowExecutionInfo().GetAssignedBuildId())
5094
s.Len(searchAttr, 2+len(extraSearchAttrBuildIds))
5095
s.Equal(worker_versioning.AssignedBuildIdSearchAttribute(expectedBuildId), searchAttr[0])
5096
s.Contains(searchAttr, worker_versioning.VersionedBuildIdSearchAttribute(expectedBuildId))
5097
} else {
5098
s.Empty(dw.GetWorkflowExecutionInfo().GetAssignedBuildId()) //nolint:staticcheck
5099
if expectedStampBuildId != "" {
5100
s.Len(searchAttr, 1+len(extraSearchAttrBuildIds))
5101
s.Contains(searchAttr, worker_versioning.VersionedBuildIdSearchAttribute(expectedBuildId))
5102
} else {
5103
s.Empty(searchAttr)
5104
}
5105
}
5106
}
5107
s.Equal(expectedInheritedBuildId, dw.GetWorkflowExecutionInfo().GetInheritedBuildId())
5108
for _, b := range extraSearchAttrBuildIds {
5109
if expectedBuildId == "" {
5110
s.Contains(searchAttr, worker_versioning.UnversionedBuildIdSearchAttribute(b))
5111
} else {
5112
s.Contains(searchAttr, worker_versioning.VersionedBuildIdSearchAttribute(b))
5113
}
5114
}
5115
}
5116
5117
func (s *VersioningIntegSuite) validateWorkflowEventsVersionStamps(
5118
env *testcore.TestEnv,
5119
wfID, runID string,
5120
expectedBuildIds []string,
5121
expectedInheritedBuildId string,
5122
) {
5123
wh := env.SdkClient().GetWorkflowHistory(s.Context(), wfID, runID, false, enumspb.HISTORY_EVENT_FILTER_TYPE_ALL_EVENT)
5124
counter := 0
5125
checkedInheritedBuildId := false
5126
for wh.HasNext() {
5127
he, err := wh.Next()
5128
s.NoError(err)
5129
if !checkedInheritedBuildId {
5130
// first event
5131
checkedInheritedBuildId = true
5132
s.Equal(expectedInheritedBuildId, he.GetWorkflowExecutionStartedEventAttributes().GetInheritedBuildId())
5133
}
5134
var taskStartedStamp *commonpb.WorkerVersionStamp
5135
if activityStarted := he.GetActivityTaskStartedEventAttributes(); activityStarted != nil {
5136
taskStartedStamp = activityStarted.GetWorkerVersion()
5137
} else if wfStarted := he.GetWorkflowTaskStartedEventAttributes(); wfStarted != nil {
5138
taskStartedStamp = wfStarted.GetWorkerVersion()
5139
}
5140
if taskStartedStamp != nil {
5141
if counter >= len(expectedBuildIds) {
5142
s.Fail("found more task started events than expected")
5143
}
5144
expected := expectedBuildIds[counter]
5145
if expected == "" {
5146
s.False(taskStartedStamp.GetUseVersioning())
5147
} else {
5148
s.True(taskStartedStamp.GetUseVersioning())
5149
s.Equal(expected, taskStartedStamp.GetBuildId())
5150
}
5151
counter++
5152
}
5153
}
5154
if counter != len(expectedBuildIds) {
5155
s.Fail("found less task started events than expected")
5156
}
5157
}