go.temporal.io/server/tests/archival_test.go
582 LOC · 0 covered · 582 uncovered · 0 ranges · 0 concepts · 0 introducers · 0 tests
1
package tests
2
3
import (
4
"bytes"
5
"context"
6
"encoding/binary"
7
"fmt"
8
"strconv"
9
"sync/atomic"
10
"testing"
11
"time"
12
13
"github.com/google/uuid"
14
commandpb "go.temporal.io/api/command/v1"
15
commonpb "go.temporal.io/api/common/v1"
16
enumspb "go.temporal.io/api/enums/v1"
17
taskqueuepb "go.temporal.io/api/taskqueue/v1"
18
workflowpb "go.temporal.io/api/workflow/v1"
19
"go.temporal.io/api/workflowservice/v1"
20
"go.temporal.io/server/api/adminservice/v1"
21
archiverspb "go.temporal.io/server/api/archiver/v1"
22
"go.temporal.io/server/chasm"
23
"go.temporal.io/server/common"
24
"go.temporal.io/server/common/archiver"
25
"go.temporal.io/server/common/archiver/filestore"
26
"go.temporal.io/server/common/archiver/provider"
27
"go.temporal.io/server/common/config"
28
"go.temporal.io/server/common/convert"
29
"go.temporal.io/server/common/dynamicconfig"
30
"go.temporal.io/server/common/log"
31
"go.temporal.io/server/common/log/tag"
32
"go.temporal.io/server/common/metrics"
33
"go.temporal.io/server/common/namespace"
34
"go.temporal.io/server/common/payloads"
35
"go.temporal.io/server/common/persistence"
36
"go.temporal.io/server/common/persistence/versionhistory"
37
"go.temporal.io/server/common/searchattribute"
38
"go.temporal.io/server/common/testing/parallelsuite"
39
"go.temporal.io/server/common/testing/protoassert"
40
"go.temporal.io/server/tests/testcore"
41
"google.golang.org/protobuf/types/known/durationpb"
42
)
43
44
const (
45
// Custom scheme for testing custom archiver implementation
46
customArchiverScheme = "customtest"
47
)
48
49
type (
50
ArchivalSuite struct {
51
parallelsuite.Suite[*ArchivalSuite]
52
}
53
54
archivalTestEnv struct {
55
*testcore.TestEnv
56
archivalNamespace namespace.Name
57
archivalNamespaceID namespace.ID
58
59
// Namespace for testing custom archiver
60
customArchiverNamespace namespace.Name
61
customArchiverNamespaceID namespace.ID
62
63
// Counters to verify custom archivers are being called
64
customHistoryArchiveCalled atomic.Int32
65
customVisibilityArchiveCalled atomic.Int32
66
67
archiverProvider provider.ArchiverProvider
68
historyURI string
69
visibilityURI string
70
}
71
72
archivalWorkflowInfo struct {
73
execution *commonpb.WorkflowExecution
74
branchToken []byte
75
}
76
77
// customHistoryArchiver wraps a built-in history archiver and tracks Archive calls
78
customHistoryArchiver struct {
79
counter *atomic.Int32
80
}
81
82
// customVisibilityArchiver wraps a built-in visibility archiver and tracks Archive calls
83
customVisibilityArchiver struct {
84
counter *atomic.Int32
85
}
86
)
87
88
// customHistoryArchiver method implementations
89
func (c *customHistoryArchiver) Archive(ctx context.Context, uri archiver.URI, request *archiver.ArchiveHistoryRequest, opts ...archiver.ArchiveOption) error {
90
c.counter.Add(1)
91
return nil
92
}
93
94
func (c *customHistoryArchiver) Get(ctx context.Context, uri archiver.URI, request *archiver.GetHistoryRequest) (*archiver.GetHistoryResponse, error) {
95
return nil, nil
96
}
97
98
func (c *customHistoryArchiver) ValidateURI(uri archiver.URI) error {
99
return nil
100
}
101
102
// customVisibilityArchiver method implementations
103
func (c *customVisibilityArchiver) Archive(ctx context.Context, uri archiver.URI, request *archiverspb.VisibilityRecord, opts ...archiver.ArchiveOption) error {
104
c.counter.Add(1)
105
return nil
106
}
107
108
func (c *customVisibilityArchiver) Query(ctx context.Context, uri archiver.URI, request *archiver.QueryVisibilityRequest, saTypeMap searchattribute.NameTypeMap) (*archiver.QueryVisibilityResponse, error) {
109
return nil, nil
110
}
111
112
func (c *customVisibilityArchiver) ValidateURI(uri archiver.URI) error {
113
return nil
114
}
115
116
func TestArchivalSuite(t *testing.T) {
117
parallelsuite.Run(t, &ArchivalSuite{})
118
}
119
120
func (s *ArchivalSuite) newTestEnv() *archivalTestEnv {
121
cfg := &config.FilestoreArchiver{FileMode: "0666", DirMode: "0766"}
122
historyProvider := &config.HistoryArchiverProvider{Filestore: cfg}
123
visibilityProvider := &config.VisibilityArchiverProvider{Filestore: cfg}
124
ae := &archivalTestEnv{
125
historyURI: filestore.URIScheme + "://" + s.T().TempDir(),
126
visibilityURI: filestore.URIScheme + "://" + s.T().TempDir(),
127
}
128
129
// Create custom history archiver factory for custom scheme
130
customHistoryArchiverFactory := provider.CustomHistoryArchiverFactoryFunc(
131
func(params provider.NewCustomHistoryArchiverParams) (archiver.HistoryArchiver, error) {
132
// Only handle custom scheme, return ErrUnknownScheme for others (including filestore)
133
if params.Scheme != customArchiverScheme {
134
return nil, provider.ErrUnknownScheme
135
}
136
// Return a wrapper that delegates to filestore but tracks Archive calls
137
return &customHistoryArchiver{
138
counter: &ae.customHistoryArchiveCalled,
139
}, nil
140
},
141
)
142
143
// Create custom visibility archiver factory for custom scheme
144
customVisibilityArchiverFactory := provider.CustomVisibilityArchiverFactoryFunc(
145
func(params provider.NewCustomVisibilityArchiverParams) (archiver.VisibilityArchiver, error) {
146
// Only handle custom scheme, return ErrUnknownScheme for others (including filestore)
147
if params.Scheme != customArchiverScheme {
148
return nil, provider.ErrUnknownScheme
149
}
150
// Return a wrapper that delegates to filestore but tracks Archive calls
151
return &customVisibilityArchiver{
152
counter: &ae.customVisibilityArchiveCalled,
153
}, nil
154
},
155
)
156
157
ae.TestEnv = testcore.NewEnv(s.T(),
158
testcore.WithDynamicConfig(dynamicconfig.ArchivalProcessorArchiveDelay, time.Duration(0)),
159
testcore.WithArchival(),
160
testcore.WithCustomArchivers(customHistoryArchiverFactory, customVisibilityArchiverFactory),
161
)
162
ae.archiverProvider = provider.NewArchiverProvider(
163
historyProvider,
164
visibilityProvider,
165
customHistoryArchiverFactory,
166
customVisibilityArchiverFactory,
167
ae.GetTestCluster().ExecutionManager(),
168
log.NewNoopLogger(),
169
metrics.NoopMetricsHandler,
170
)
171
172
var err error
173
174
// Register namespace using built-in filestore archiver
175
ae.archivalNamespace = namespace.Name(testcore.RandomizeStr("archival-enabled-namespace"))
176
ae.archivalNamespaceID, err = ae.RegisterNamespace(
177
ae.archivalNamespace,
178
0, // Archive right away.
179
enumspb.ARCHIVAL_STATE_ENABLED,
180
ae.historyURI,
181
ae.visibilityURI,
182
)
183
s.NoError(err)
184
185
// Register namespace using custom archiver with custom scheme
186
ae.customArchiverNamespace = namespace.Name(testcore.RandomizeStr("custom-archiver-namespace"))
187
customHistoryURI := customArchiverScheme + "://custom-history-archiver"
188
customVisibilityURI := customArchiverScheme + "://custom-visibility-archiver"
189
ae.customArchiverNamespaceID, err = ae.RegisterNamespace(
190
ae.customArchiverNamespace,
191
0, // Archive right away.
192
enumspb.ARCHIVAL_STATE_ENABLED,
193
customHistoryURI,
194
customVisibilityURI,
195
)
196
s.NoError(err)
197
198
return ae
199
}
200
201
func (s *ArchivalSuite) TestArchival_TimerQueueProcessor() {
202
env := s.newTestEnv()
203
204
workflowID := "archival-timer-queue-processor-workflow-id"
205
workflowType := "archival-timer-queue-processor-type"
206
taskQueue := "archival-timer-queue-processor-task-queue"
207
numActivities := 1
208
numRuns := 1
209
workflowInfo := s.startAndFinishWorkflow(env, workflowID, workflowType, taskQueue, env.archivalNamespace, numActivities, numRuns)[0]
210
211
s.workflowIsArchived(env, env.archivalNamespaceID, workflowInfo.execution)
212
s.historyIsDeleted(env, workflowInfo)
213
s.mutableStateIsDeleted(env, env.archivalNamespaceID, workflowInfo.execution)
214
}
215
216
func (s *ArchivalSuite) TestArchival_ContinueAsNew() {
217
env := s.newTestEnv()
218
219
workflowID := "archival-continueAsNew-workflow-id"
220
workflowType := "archival-continueAsNew-workflow-type"
221
taskQueue := "archival-continueAsNew-task-queue"
222
numActivities := 1
223
numRuns := 5
224
workflowInfos := s.startAndFinishWorkflow(env, workflowID, workflowType, taskQueue, env.archivalNamespace, numActivities, numRuns)
225
226
for _, workflowInfo := range workflowInfos {
227
s.workflowIsArchived(env, env.archivalNamespaceID, workflowInfo.execution)
228
s.historyIsDeleted(env, workflowInfo)
229
s.mutableStateIsDeleted(env, env.archivalNamespaceID, workflowInfo.execution)
230
}
231
}
232
233
func (s *ArchivalSuite) TestArchival_ArchiverWorker() {
234
// s.T().SkipNow() // flaky test, skip for now, will reimplement archival feature.
235
236
env := s.newTestEnv()
237
238
workflowID := "archival-archiver-worker-workflow-id"
239
workflowType := "archival-archiver-worker-workflow-type"
240
taskQueue := "archival-archiver-worker-task-queue"
241
numActivities := 10
242
workflowInfo := s.startAndFinishWorkflow(env, workflowID, workflowType, taskQueue, env.archivalNamespace, numActivities, 1)[0]
243
244
s.workflowIsArchived(env, env.archivalNamespaceID, workflowInfo.execution)
245
s.historyIsDeleted(env, workflowInfo)
246
s.mutableStateIsDeleted(env, env.archivalNamespaceID, workflowInfo.execution)
247
}
248
249
func (s *ArchivalSuite) TestVisibilityArchival() {
250
env := s.newTestEnv()
251
252
workflowID := "archival-visibility-workflow-id"
253
workflowType := "archival-visibility-workflow-type"
254
taskQueue := "archival-visibility-task-queue"
255
numActivities := 3
256
numRuns := 5
257
startTime := time.Now().UnixNano()
258
s.startAndFinishWorkflow(env, workflowID, workflowType, taskQueue, env.archivalNamespace, numActivities, numRuns)
259
s.startAndFinishWorkflow(env, "some other workflowID", "some other workflow type", taskQueue, env.archivalNamespace, numActivities, numRuns)
260
endTime := time.Now().UnixNano()
261
262
var executions []*workflowpb.WorkflowExecutionInfo
263
264
s.Eventually(func() bool {
265
request := &workflowservice.ListArchivedWorkflowExecutionsRequest{
266
Namespace: env.archivalNamespace.String(),
267
PageSize: 2,
268
Query: fmt.Sprintf("CloseTime >= %v and CloseTime <= %v and WorkflowType = '%s'", startTime, endTime, workflowType),
269
}
270
for len(executions) == 0 || request.NextPageToken != nil {
271
response, err := env.FrontendClient().ListArchivedWorkflowExecutions(s.Context(), request)
272
s.NoError(err)
273
s.NotNil(response)
274
executions = append(executions, response.GetExecutions()...)
275
request.NextPageToken = response.NextPageToken
276
}
277
if len(executions) == numRuns {
278
return true
279
}
280
return false
281
}, 20*time.Second, 500*time.Millisecond)
282
283
for _, execution := range executions {
284
s.Equal(workflowID, execution.GetExecution().GetWorkflowId())
285
s.Equal(workflowType, execution.GetType().GetName())
286
s.NotZero(execution.StartTime)
287
s.NotZero(execution.ExecutionTime)
288
s.NotZero(execution.CloseTime)
289
s.NotZero(execution.ExecutionDuration)
290
s.Equal(
291
execution.CloseTime.AsTime().Sub(execution.ExecutionTime.AsTime()),
292
execution.ExecutionDuration.AsDuration(),
293
)
294
}
295
}
296
297
func (s *ArchivalSuite) TestCustomArchiver() {
298
env := s.newTestEnv()
299
300
workflowID := "custom-history-archiver-workflow-id"
301
workflowType := "custom-history-archiver-type"
302
taskQueue := "custom-history-archiver-task-queue"
303
numActivities := 1
304
numRuns := 1
305
306
// Reset counter before test
307
env.customHistoryArchiveCalled.Store(0)
308
env.customVisibilityArchiveCalled.Store(0)
309
310
// Use custom archiver namespace to trigger custom archiver
311
s.startAndFinishWorkflow(env, workflowID, workflowType, taskQueue, env.customArchiverNamespace, numActivities, numRuns)
312
313
// Verify custom archiver's Archive method was called at least once
314
s.Eventually(func() bool {
315
called := env.customHistoryArchiveCalled.Load()
316
return called > 0
317
}, 10*time.Second, 500*time.Millisecond, "Custom history archiver Archive method should have been called")
318
s.Eventually(func() bool {
319
called := env.customVisibilityArchiveCalled.Load()
320
return called > 0
321
}, 10*time.Second, 500*time.Millisecond, "Custom visibility archiver Archive method should have been called")
322
}
323
324
// workflowIsArchived asserts that both the workflow history and workflow visibility are archived.
325
func (s *ArchivalSuite) workflowIsArchived(env *archivalTestEnv, namespaceID namespace.ID, execution *commonpb.WorkflowExecution) {
326
historyURI, err := archiver.NewURI(env.historyURI)
327
s.NoError(err)
328
historyArchiver, err := env.archiverProvider.GetHistoryArchiver(
329
historyURI.Scheme(),
330
)
331
s.NoError(err)
332
333
visibilityURI, err := archiver.NewURI(env.visibilityURI)
334
s.NoError(err)
335
visibilityArchiver, err := env.archiverProvider.GetVisibilityArchiver(
336
visibilityURI.Scheme(),
337
)
338
s.NoError(err)
339
340
s.Eventually(func() bool {
341
var historyResponse *archiver.GetHistoryResponse
342
historyResponse, err = historyArchiver.Get(s.Context(), historyURI, &archiver.GetHistoryRequest{
343
NamespaceID: namespaceID.String(),
344
WorkflowID: execution.GetWorkflowId(),
345
RunID: execution.GetRunId(),
346
PageSize: 1,
347
})
348
if err != nil {
349
return false
350
}
351
if len(historyResponse.HistoryBatches) == 0 {
352
return false
353
}
354
var visibilityResponse *archiver.QueryVisibilityResponse
355
visibilityResponse, err = visibilityArchiver.Query(
356
s.Context(),
357
visibilityURI,
358
&archiver.QueryVisibilityRequest{
359
NamespaceID: namespaceID.String(),
360
PageSize: 1,
361
Query: fmt.Sprintf(
362
"WorkflowId = '%s' and RunId = '%s'",
363
execution.GetWorkflowId(),
364
execution.GetRunId(),
365
),
366
},
367
searchattribute.NameTypeMap{},
368
)
369
if err != nil {
370
return false
371
}
372
if len(visibilityResponse.Executions) > 0 {
373
return true
374
}
375
return false
376
}, 20*time.Second, 500*time.Millisecond)
377
}
378
379
func (s *ArchivalSuite) historyIsDeleted(env *archivalTestEnv, workflowInfo archivalWorkflowInfo) {
380
shardID := common.WorkflowIDToHistoryShard(
381
env.archivalNamespaceID.String(),
382
workflowInfo.execution.WorkflowId,
383
env.GetTestClusterConfig().HistoryConfig.NumHistoryShards,
384
)
385
386
s.Eventually(func() bool {
387
_, err := env.GetTestCluster().TestBase().ExecutionManager.ReadHistoryBranch(
388
s.Context(),
389
&persistence.ReadHistoryBranchRequest{
390
ShardID: shardID,
391
BranchToken: workflowInfo.branchToken,
392
MinEventID: common.FirstEventID,
393
MaxEventID: common.EndEventID,
394
PageSize: 1,
395
NextPageToken: nil,
396
},
397
)
398
if common.IsNotFoundError(err) {
399
return true
400
}
401
s.NoError(err)
402
return false
403
}, 20*time.Second, 500*time.Millisecond)
404
}
405
406
func (s *ArchivalSuite) mutableStateIsDeleted(env *archivalTestEnv, namespaceID namespace.ID, execution *commonpb.WorkflowExecution) {
407
shardID := common.WorkflowIDToHistoryShard(namespaceID.String(), execution.GetWorkflowId(),
408
env.GetTestClusterConfig().HistoryConfig.NumHistoryShards)
409
request := &persistence.GetWorkflowExecutionRequest{
410
ShardID: shardID,
411
NamespaceID: namespaceID.String(),
412
WorkflowID: execution.WorkflowId,
413
RunID: execution.RunId,
414
ArchetypeID: chasm.WorkflowArchetypeID,
415
}
416
417
s.Eventually(func() bool {
418
_, err := env.GetTestCluster().TestBase().ExecutionManager.GetWorkflowExecution(s.Context(), request)
419
if common.IsNotFoundError(err) {
420
return true
421
}
422
s.NoError(err)
423
return false
424
}, 20*time.Second, 500*time.Millisecond)
425
}
426
427
func (s *ArchivalSuite) startAndFinishWorkflow(
428
env *archivalTestEnv,
429
id, wt, tq string,
430
nsName namespace.Name,
431
numActivities, numRuns int,
432
) []archivalWorkflowInfo {
433
identity := "worker1"
434
activityName := "activity_type1"
435
workflowType := &commonpb.WorkflowType{Name: wt}
436
taskQueue := &taskqueuepb.TaskQueue{Name: tq, Kind: enumspb.TASK_QUEUE_KIND_NORMAL}
437
request := &workflowservice.StartWorkflowExecutionRequest{
438
RequestId: uuid.NewString(),
439
Namespace: nsName.String(),
440
WorkflowId: id,
441
WorkflowType: workflowType,
442
TaskQueue: taskQueue,
443
Input: nil,
444
WorkflowRunTimeout: durationpb.New(100 * time.Second),
445
WorkflowTaskTimeout: durationpb.New(1 * time.Second),
446
Identity: identity,
447
}
448
startResp, err := env.FrontendClient().StartWorkflowExecution(s.Context(), request)
449
s.NoError(err)
450
env.Logger.Info("StartWorkflowExecution", tag.WorkflowRunID(startResp.RunId))
451
workflowInfos := make([]archivalWorkflowInfo, numRuns)
452
453
workflowComplete := false
454
activityCount := int32(numActivities)
455
activityCounter := int32(0)
456
expectedActivityID := int32(1)
457
runCounter := 1
458
459
wtHandler := func(task *workflowservice.PollWorkflowTaskQueueResponse) ([]*commandpb.Command, error) {
460
branchToken, err := s.getBranchToken(env, nsName, task.WorkflowExecution)
461
s.NoError(err)
462
463
workflowInfos[runCounter-1] = archivalWorkflowInfo{
464
execution: task.WorkflowExecution,
465
branchToken: branchToken,
466
}
467
468
if activityCounter < activityCount {
469
activityCounter++
470
buf := new(bytes.Buffer)
471
s.NoError(binary.Write(buf, binary.LittleEndian, activityCounter))
472
return []*commandpb.Command{{
473
CommandType: enumspb.COMMAND_TYPE_SCHEDULE_ACTIVITY_TASK,
474
Attributes: &commandpb.Command_ScheduleActivityTaskCommandAttributes{ScheduleActivityTaskCommandAttributes: &commandpb.ScheduleActivityTaskCommandAttributes{
475
ActivityId: convert.Int32ToString(activityCounter),
476
ActivityType: &commonpb.ActivityType{Name: activityName},
477
TaskQueue: &taskqueuepb.TaskQueue{Name: tq, Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
478
Input: payloads.EncodeBytes(buf.Bytes()),
479
ScheduleToCloseTimeout: durationpb.New(100 * time.Second),
480
ScheduleToStartTimeout: durationpb.New(10 * time.Second),
481
StartToCloseTimeout: durationpb.New(50 * time.Second),
482
HeartbeatTimeout: durationpb.New(5 * time.Second),
483
}},
484
}}, nil
485
}
486
487
if runCounter < numRuns {
488
activityCounter = int32(0)
489
expectedActivityID = int32(1)
490
runCounter++
491
return []*commandpb.Command{{
492
CommandType: enumspb.COMMAND_TYPE_CONTINUE_AS_NEW_WORKFLOW_EXECUTION,
493
Attributes: &commandpb.Command_ContinueAsNewWorkflowExecutionCommandAttributes{ContinueAsNewWorkflowExecutionCommandAttributes: &commandpb.ContinueAsNewWorkflowExecutionCommandAttributes{
494
WorkflowType: workflowType,
495
TaskQueue: &taskqueuepb.TaskQueue{Name: tq, Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
496
Input: nil,
497
WorkflowRunTimeout: durationpb.New(100 * time.Second),
498
WorkflowTaskTimeout: durationpb.New(1 * time.Second),
499
}},
500
}}, nil
501
}
502
503
workflowComplete = true
504
return []*commandpb.Command{{
505
CommandType: enumspb.COMMAND_TYPE_COMPLETE_WORKFLOW_EXECUTION,
506
Attributes: &commandpb.Command_CompleteWorkflowExecutionCommandAttributes{CompleteWorkflowExecutionCommandAttributes: &commandpb.CompleteWorkflowExecutionCommandAttributes{
507
Result: payloads.EncodeString("Done"),
508
}},
509
}}, nil
510
}
511
512
atHandler := func(task *workflowservice.PollActivityTaskQueueResponse) (*commonpb.Payloads, bool, error) {
513
protoassert.ProtoEqual(s.T(), workflowInfos[runCounter-1].execution, task.WorkflowExecution)
514
s.Equal(activityName, task.ActivityType.Name)
515
currentActivityId, _ := strconv.Atoi(task.ActivityId)
516
s.Equal(int(expectedActivityID), currentActivityId)
517
var inputBytes []byte
518
s.NoError(payloads.Decode(task.Input, &inputBytes))
519
s.Equal(expectedActivityID, int32(binary.LittleEndian.Uint32(inputBytes)))
520
expectedActivityID++
521
return payloads.EncodeString("Activity Result"), false, nil
522
}
523
524
poller := &testcore.TaskPoller{
525
Client: env.FrontendClient(),
526
Namespace: nsName.String(),
527
TaskQueue: taskQueue,
528
Identity: identity,
529
WorkflowTaskHandler: wtHandler,
530
ActivityTaskHandler: atHandler,
531
Logger: env.Logger,
532
T: s.T(),
533
}
534
for range numRuns {
535
for i := range numActivities {
536
_, err := poller.PollAndProcessWorkflowTask()
537
env.Logger.Info("PollAndProcessWorkflowTask", tag.Error(err))
538
s.NoError(err)
539
if i%2 == 0 {
540
err = poller.PollAndProcessActivityTask(false)
541
} else { // just for testing respondActivityTaskCompleteByID
542
err = poller.PollAndProcessActivityTaskWithID(false)
543
}
544
env.Logger.Info("PollAndProcessActivityTask", tag.Error(err))
545
s.NoError(err)
546
}
547
548
_, err = poller.PollAndProcessWorkflowTask(testcore.WithDumpHistory)
549
s.NoError(err)
550
}
551
552
s.True(workflowComplete)
553
for run := 1; run < numRuns; run++ {
554
s.NotEqual(workflowInfos[run-1].execution, workflowInfos[run].execution)
555
s.NotEqual(workflowInfos[run-1].branchToken, workflowInfos[run].branchToken)
556
}
557
return workflowInfos
558
}
559
560
func (s *ArchivalSuite) getBranchToken(
561
env *archivalTestEnv,
562
nsName namespace.Name,
563
execution *commonpb.WorkflowExecution,
564
) ([]byte, error) {
565
566
descResp, err := env.AdminClient().DescribeMutableState(s.Context(), &adminservice.DescribeMutableStateRequest{
567
Namespace: nsName.String(),
568
Execution: execution,
569
Archetype: chasm.WorkflowArchetype,
570
})
571
if err != nil {
572
return nil, err
573
}
574
575
versionHistories := descResp.CacheMutableState.ExecutionInfo.VersionHistories
576
currentVersionHistory, err := versionhistory.GetCurrentVersionHistory(versionHistories)
577
if err != nil {
578
return nil, err
579
}
580
581
return currentVersionHistory.GetBranchToken(), nil
582
}