go.temporal.io/server/tests/callbacks_test.go
800 LOC · 0 covered · 800 uncovered · 0 ranges · 0 concepts · 0 introducers · 0 tests
1
package tests
2
3
import (
4
"context"
5
"errors"
6
"net/http/httptest"
7
"testing"
8
"time"
9
10
"github.com/google/uuid"
11
"github.com/nexus-rpc/sdk-go/nexus"
12
"github.com/stretchr/testify/require"
13
commonpb "go.temporal.io/api/common/v1"
14
enumspb "go.temporal.io/api/enums/v1"
15
schedulepb "go.temporal.io/api/schedule/v1"
16
"go.temporal.io/api/serviceerror"
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/sdk/workflow"
21
"go.temporal.io/server/chasm"
22
"go.temporal.io/server/chasm/lib/callback"
23
"go.temporal.io/server/common/dynamicconfig"
24
commonnexus "go.temporal.io/server/common/nexus"
25
"go.temporal.io/server/common/nexus/nexusrpc"
26
"go.temporal.io/server/common/testing/await"
27
"go.temporal.io/server/common/testing/parallelsuite"
28
"go.temporal.io/server/common/testing/protoassert"
29
"go.temporal.io/server/common/testing/protorequire"
30
"go.temporal.io/server/tests/testcore"
31
"google.golang.org/protobuf/proto"
32
"google.golang.org/protobuf/types/known/durationpb"
33
)
34
35
type completionHandler struct {
36
requestCh chan *nexusrpc.CompletionRequest
37
requestCompleteCh chan error
38
}
39
40
func (h *completionHandler) CompleteOperation(ctx context.Context, request *nexusrpc.CompletionRequest) error {
41
h.requestCh <- request
42
return <-h.requestCompleteCh
43
}
44
45
type CallbacksSuite struct {
46
parallelsuite.Suite[*CallbacksSuite]
47
}
48
49
func TestCallbacksSuiteHSM(t *testing.T) {
50
parallelsuite.Run(t, &CallbacksSuite{}, []testcore.TestOption{})
51
}
52
53
func TestCallbacksSuiteCHASM(t *testing.T) {
54
parallelsuite.Run(t, &CallbacksSuite{}, []testcore.TestOption{
55
testcore.WithDynamicConfig(dynamicconfig.EnableChasm, true),
56
testcore.WithDynamicConfig(dynamicconfig.EnableCHASMCallbacks, true),
57
})
58
}
59
60
func (s *CallbacksSuite) runNexusCompletionHTTPServer(t *testing.T, h *completionHandler) string {
61
hh := nexusrpc.NewCompletionHTTPHandler(nexusrpc.CompletionHandlerOptions{Handler: h})
62
srv := httptest.NewServer(hh)
63
t.Cleanup(func() {
64
srv.Close()
65
})
66
return srv.URL
67
}
68
69
func (s *CallbacksSuite) newTestEnv(opts ...testcore.TestOption) *testcore.TestEnv {
70
env := testcore.NewEnv(s.T(), opts...)
71
env.OverrideDynamicConfig(
72
callback.AllowedAddresses,
73
[]any{map[string]any{"Pattern": "*", "AllowInsecure": true}},
74
)
75
return env
76
}
77
78
func (s *CallbacksSuite) TestScheduledCallbackTokenMigration_LegacyWriteEnvelopeRead(opts []testcore.TestOption) {
79
testOpts := append([]testcore.TestOption{}, opts...)
80
testOpts = append(
81
testOpts,
82
scheduleCommonOpts(s.T())...,
83
)
84
testOpts = append(
85
testOpts,
86
testcore.WithDynamicConfig(dynamicconfig.EnableCHASMCallbacks, true),
87
testcore.WithDynamicConfig(callback.EncodeInternalTokenWithEnvelope, false),
88
)
89
env := newScheduleEnv(s.T(), testOpts...)
90
91
ctx := s.Context()
92
sid := testcore.RandomizeStr("sched-token-migration")
93
wid := testcore.RandomizeStr("sched-token-migration-wf")
94
wt := testcore.RandomizeStr("sched-token-migration-wt")
95
96
env.SdkWorker().RegisterWorkflowWithOptions(func(ctx workflow.Context) error {
97
workflow.GetSignalChannel(ctx, "continue").Receive(ctx, nil)
98
return nil
99
}, workflow.RegisterOptions{Name: wt})
100
101
schedule := &schedulepb.Schedule{
102
Spec: &schedulepb.ScheduleSpec{
103
Interval: []*schedulepb.IntervalSpec{{Interval: durationpb.New(1 * time.Second)}},
104
},
105
Policies: &schedulepb.SchedulePolicies{OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_SKIP},
106
Action: &schedulepb.ScheduleAction{
107
Action: &schedulepb.ScheduleAction_StartWorkflow{
108
StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
109
WorkflowId: wid,
110
WorkflowType: &commonpb.WorkflowType{Name: wt},
111
TaskQueue: &taskqueuepb.TaskQueue{Name: env.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
112
},
113
},
114
},
115
}
116
_, err := env.FrontendClient().CreateSchedule(chasmContextFactory(ctx), &workflowservice.CreateScheduleRequest{
117
Namespace: env.Namespace().String(),
118
ScheduleId: sid,
119
Schedule: schedule,
120
Identity: "test",
121
RequestId: uuid.NewString(),
122
})
123
s.NoError(err)
124
125
var startedWFID string
126
await.RequireTruef(s.T(), func() bool {
127
desc, descErr := env.FrontendClient().DescribeSchedule(chasmContextFactory(ctx), &workflowservice.DescribeScheduleRequest{
128
Namespace: env.Namespace().String(),
129
ScheduleId: sid,
130
})
131
if descErr != nil {
132
return false
133
}
134
for _, a := range desc.GetInfo().GetRecentActions() {
135
if wfid := a.GetStartWorkflowResult().GetWorkflowId(); wfid != "" {
136
startedWFID = wfid
137
return true
138
}
139
}
140
return false
141
}, 15*time.Second, 200*time.Millisecond, "scheduler should start a scheduled action")
142
s.NotEmpty(startedWFID)
143
144
var token string
145
for _, e := range env.GetHistory(env.Namespace().String(), &commonpb.WorkflowExecution{WorkflowId: startedWFID}) {
146
for _, cb := range e.GetWorkflowExecutionStartedEventAttributes().GetCompletionCallbacks() {
147
if cb.GetNexus().GetUrl() == chasm.NexusCompletionHandlerURL {
148
token = nexus.Header(cb.GetNexus().GetHeader()).Get(commonnexus.CallbackTokenHeader)
149
}
150
}
151
}
152
s.NotEmpty(token, "scheduled workflow must carry an internal completion callback token")
153
_, reqID, decErr := chasm.UnpackNexusCallbackToken(token)
154
s.NoError(decErr)
155
s.Empty(reqID, "gate OFF must write a legacy token with no embedded request ID")
156
157
env.OverrideDynamicConfig(callback.EncodeInternalTokenWithEnvelope, true)
158
159
_, err = env.FrontendClient().SignalWorkflowExecution(ctx, &workflowservice.SignalWorkflowExecutionRequest{
160
Namespace: env.Namespace().String(),
161
WorkflowExecution: &commonpb.WorkflowExecution{WorkflowId: startedWFID},
162
SignalName: "continue",
163
})
164
s.NoError(err)
165
166
await.RequireTruef(s.T(), func() bool {
167
desc, descErr := env.FrontendClient().DescribeSchedule(chasmContextFactory(ctx), &workflowservice.DescribeScheduleRequest{
168
Namespace: env.Namespace().String(),
169
ScheduleId: sid,
170
})
171
if descErr != nil {
172
return false
173
}
174
for _, a := range desc.GetInfo().GetRecentActions() {
175
if a.GetStartWorkflowResult().GetWorkflowId() == startedWFID &&
176
a.GetStartWorkflowStatus() == enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED {
177
return true
178
}
179
}
180
return false
181
}, 20*time.Second, 200*time.Millisecond,
182
"scheduler must observe completion of the legacy-token action after enabling the envelope gate")
183
}
184
185
func (s *CallbacksSuite) TestWorkflowCallbacks_InvalidArgument(opts []testcore.TestOption) {
186
workflowType := "test"
187
188
cases := []struct {
189
name string
190
urls []string
191
header map[string]string
192
message string
193
}{
194
{
195
name: "invalid-scheme",
196
urls: []string{"invalid"},
197
message: "invalid url: unknown scheme: invalid",
198
},
199
{
200
name: "url-length-too-long",
201
urls: []string{"http://some-very-very-very-very-very-very-very-long-url"},
202
message: "invalid url: url length longer than max length allowed of 50",
203
},
204
{
205
name: "header-size-too-large",
206
urls: []string{"http://some-ignored-address"},
207
header: map[string]string{"too": "long"},
208
message: "invalid header: header size longer than max allowed size of 6",
209
},
210
{
211
name: "too many callbacks",
212
urls: []string{"http://url-1", "http://url-2", "http://url-3"},
213
message: "cannot attach more than 2 callbacks to an execution",
214
},
215
{
216
name: "url not configured",
217
urls: []string{"http://some-unconfigured-address"},
218
message: "invalid url: url does not match any configured callback address: http://some-unconfigured-address",
219
},
220
{
221
name: "https required",
222
urls: []string{"http://some-secure-address"},
223
message: "invalid url: callback address does not allow insecure connections: http://some-secure-address",
224
},
225
}
226
227
for _, tc := range cases {
228
s.Run(tc.name, func(s *CallbacksSuite) {
229
env := testcore.NewEnv(s.T(), opts...)
230
env.OverrideDynamicConfig(dynamicconfig.FrontendCallbackURLMaxLength, 50)
231
env.OverrideDynamicConfig(dynamicconfig.FrontendCallbackHeaderMaxSize, 6)
232
env.OverrideDynamicConfig(dynamicconfig.MaxCallbacksPerWorkflow, 2)
233
env.OverrideDynamicConfig(callback.MaxPerExecution, 2)
234
env.OverrideDynamicConfig(
235
callback.AllowedAddresses,
236
[]any{map[string]any{"Pattern": "some-ignored-address", "AllowInsecure": true}, map[string]any{"Pattern": "some-secure-address", "AllowInsecure": false}},
237
)
238
239
taskQueue := testcore.RandomizeStr(s.T().Name())
240
241
cbs := make([]*commonpb.Callback, 0, len(tc.urls))
242
for _, url := range tc.urls {
243
cbs = append(cbs, &commonpb.Callback{
244
Variant: &commonpb.Callback_Nexus_{
245
Nexus: &commonpb.Callback_Nexus{
246
Url: url,
247
Header: tc.header,
248
},
249
},
250
})
251
}
252
request := &workflowservice.StartWorkflowExecutionRequest{
253
RequestId: uuid.NewString(),
254
Namespace: env.Namespace().String(),
255
WorkflowId: testcore.RandomizeStr(s.T().Name()),
256
WorkflowType: &commonpb.WorkflowType{Name: workflowType},
257
TaskQueue: &taskqueuepb.TaskQueue{Name: taskQueue, Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
258
Input: nil,
259
WorkflowRunTimeout: durationpb.New(100 * time.Second),
260
Identity: s.T().Name(),
261
CompletionCallbacks: cbs,
262
}
263
264
_, err := env.FrontendClient().StartWorkflowExecution(s.Context(), request)
265
var invalidArgument *serviceerror.InvalidArgument
266
s.ErrorAs(err, &invalidArgument)
267
s.Equal(tc.message, err.Error())
268
})
269
}
270
}
271
272
func (s *CallbacksSuite) TestWorkflowNexusCallbacks_CarriedOver(opts []testcore.TestOption) {
273
cases := []struct {
274
name string
275
wf func(workflow.Context) (int, error)
276
runTimeout time.Duration
277
}{
278
{
279
name: "ContinueAsNew",
280
wf: func(ctx workflow.Context) (int, error) {
281
if workflow.GetInfo(ctx).ContinuedExecutionRunID == "" {
282
workflow.GetSignalChannel(ctx, "continue").Receive(ctx, nil)
283
return 0, workflow.NewContinueAsNewError(ctx, "test")
284
}
285
return 666, nil
286
},
287
runTimeout: 100 * time.Second,
288
},
289
{
290
name: "WorkflowRunTimeout",
291
wf: func(ctx workflow.Context) (int, error) {
292
info := workflow.GetInfo(ctx)
293
if info.FirstRunID == info.WorkflowExecution.RunID {
294
workflow.GetSignalChannel(ctx, "continue").Receive(ctx, nil)
295
return 0, workflow.Sleep(ctx, 10*time.Second)
296
}
297
return 666, nil
298
},
299
runTimeout: 500 * time.Millisecond,
300
},
301
{
302
name: "WorkflowFailureRetry",
303
wf: func(ctx workflow.Context) (int, error) {
304
info := workflow.GetInfo(ctx)
305
if info.FirstRunID == info.WorkflowExecution.RunID {
306
workflow.GetSignalChannel(ctx, "continue").Receive(ctx, nil)
307
return 0, errors.New("intentional workflow failure")
308
}
309
return 666, nil
310
},
311
runTimeout: 100 * time.Second,
312
},
313
}
314
315
for _, tc := range cases {
316
s.Run(tc.name, func(s *CallbacksSuite) {
317
env := s.newTestEnv(opts...)
318
319
ctx := s.Context()
320
sdkClient := env.SdkClient()
321
322
workflowType := "test"
323
workflowID := env.Tv().WorkflowID()
324
325
ch := &completionHandler{
326
requestCh: make(chan *nexusrpc.CompletionRequest, 2),
327
requestCompleteCh: make(chan error, 2),
328
}
329
defer func() {
330
close(ch.requestCh)
331
close(ch.requestCompleteCh)
332
}()
333
callbackAddress := s.runNexusCompletionHTTPServer(s.T(), ch)
334
335
env.SdkWorker().RegisterWorkflowWithOptions(tc.wf, workflow.RegisterOptions{Name: workflowType})
336
337
links := []*commonpb.Link{
338
{
339
Variant: &commonpb.Link_WorkflowEvent_{
340
WorkflowEvent: &commonpb.Link_WorkflowEvent{
341
Namespace: env.Namespace().String(),
342
WorkflowId: "some-caller-wfid-1",
343
RunId: "some-caller-runid-1",
344
},
345
},
346
},
347
{
348
Variant: &commonpb.Link_WorkflowEvent_{
349
WorkflowEvent: &commonpb.Link_WorkflowEvent{
350
Namespace: env.Namespace().String(),
351
WorkflowId: "some-caller-wfid-2",
352
RunId: "some-caller-runid-2",
353
},
354
},
355
},
356
}
357
358
cbs := []*commonpb.Callback{
359
{
360
Variant: &commonpb.Callback_Nexus_{
361
Nexus: &commonpb.Callback_Nexus{
362
Url: callbackAddress + "/cb1",
363
},
364
},
365
Links: []*commonpb.Link{links[0]},
366
},
367
{
368
Variant: &commonpb.Callback_Nexus_{
369
Nexus: &commonpb.Callback_Nexus{
370
Url: callbackAddress + "/cb2",
371
},
372
},
373
Links: []*commonpb.Link{links[1]},
374
},
375
}
376
377
request := &workflowservice.StartWorkflowExecutionRequest{
378
RequestId: uuid.NewString(),
379
Namespace: env.Namespace().String(),
380
WorkflowId: workflowID,
381
WorkflowType: &commonpb.WorkflowType{Name: workflowType},
382
TaskQueue: &taskqueuepb.TaskQueue{Name: env.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
383
Input: nil,
384
WorkflowRunTimeout: durationpb.New(tc.runTimeout),
385
Identity: s.T().Name(),
386
RetryPolicy: &commonpb.RetryPolicy{
387
InitialInterval: durationpb.New(1 * time.Second),
388
MaximumInterval: durationpb.New(1 * time.Second),
389
BackoffCoefficient: 1,
390
},
391
CompletionCallbacks: []*commonpb.Callback{cbs[0]},
392
Links: []*commonpb.Link{links[0]},
393
}
394
395
response1, err := env.FrontendClient().StartWorkflowExecution(ctx, request)
396
s.NoError(err)
397
398
workflowExecution := &commonpb.WorkflowExecution{
399
WorkflowId: workflowID,
400
RunId: response1.RunId,
401
}
402
403
// Send another request to attach callback
404
request2 := proto.Clone(request).(*workflowservice.StartWorkflowExecutionRequest)
405
request2.RequestId = uuid.NewString()
406
request2.WorkflowIdConflictPolicy = enumspb.WORKFLOW_ID_CONFLICT_POLICY_USE_EXISTING
407
request2.OnConflictOptions = &workflowpb.OnConflictOptions{
408
AttachRequestId: true,
409
AttachCompletionCallbacks: true,
410
}
411
request2.CompletionCallbacks = []*commonpb.Callback{cbs[1]}
412
request2.Links = []*commonpb.Link{links[1]}
413
414
response2, err := env.FrontendClient().StartWorkflowExecution(ctx, request2)
415
s.NoError(err)
416
s.False(response2.Started)
417
s.Equal(workflowExecution.RunId, response2.RunId)
418
419
_, err = env.FrontendClient().SignalWorkflowExecution(
420
ctx,
421
&workflowservice.SignalWorkflowExecutionRequest{
422
Namespace: env.Namespace().String(),
423
WorkflowExecution: workflowExecution,
424
SignalName: "continue",
425
},
426
)
427
s.NoError(err)
428
429
// Wait for workflow to complete.
430
run := sdkClient.GetWorkflow(ctx, workflowID, "")
431
s.NoError(run.Get(ctx, nil))
432
433
numAttempts := 2
434
for attempt := 1; attempt <= numAttempts; attempt++ {
435
for range cbs {
436
completion := <-ch.requestCh
437
s.Equal(nexus.OperationStateSucceeded, completion.State)
438
var result int
439
s.NoError(completion.Result.Consume(&result))
440
s.Equal(666, result)
441
}
442
443
for range cbs {
444
var err error
445
if attempt < numAttempts {
446
// force retry
447
err = nexus.NewHandlerErrorf(nexus.HandlerErrorTypeInternal, "intentional error")
448
}
449
ch.requestCompleteCh <- err
450
}
451
452
getHistoryResponse, err := env.FrontendClient().GetWorkflowExecutionHistory(
453
ctx,
454
&workflowservice.GetWorkflowExecutionHistoryRequest{
455
Namespace: env.Namespace().String(),
456
Execution: &commonpb.WorkflowExecution{
457
WorkflowId: workflowID,
458
},
459
MaximumPageSize: 1, // only interested in the start event
460
},
461
)
462
s.NoError(err)
463
startEvent := s.RequireHistoryEvent(
464
getHistoryResponse.History.Events,
465
enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED,
466
)
467
468
// Start event links is empty since it's deduped.
469
s.Empty(startEvent.Links)
470
startEventAttr := startEvent.GetWorkflowExecutionStartedEventAttributes()
471
s.NotNil(startEventAttr)
472
// Start event contains all callbacks attached to the first workflow.
473
s.ProtoElementsMatch(cbs, startEventAttr.CompletionCallbacks)
474
475
await.Require(s.Context(), s.T(), func(col *await.T) {
476
description, err := sdkClient.DescribeWorkflowExecution(ctx, workflowID, "")
477
require.NoError(col, err)
478
require.Len(col, description.Callbacks, len(cbs))
479
descCbs := make([]*commonpb.Callback, 0, len(description.Callbacks))
480
for _, callbackInfo := range description.Callbacks {
481
protorequire.ProtoEqual(
482
col,
483
&workflowpb.CallbackInfo_Trigger{
484
Variant: &workflowpb.CallbackInfo_Trigger_WorkflowClosed{
485
WorkflowClosed: &workflowpb.CallbackInfo_WorkflowClosed{},
486
},
487
},
488
callbackInfo.Trigger,
489
)
490
require.Equal(col, int32(attempt), callbackInfo.Attempt)
491
// Loose check to see that this is set.
492
require.Greater(
493
col,
494
callbackInfo.LastAttemptCompleteTime.AsTime(),
495
time.Now().Add(-time.Hour),
496
)
497
if attempt < numAttempts {
498
require.Equal(col, enumspb.CALLBACK_STATE_BACKING_OFF, callbackInfo.State)
499
require.Equal(
500
col,
501
"handler error (INTERNAL): intentional error",
502
callbackInfo.LastAttemptFailure.Message,
503
)
504
} else {
505
require.Equal(col, enumspb.CALLBACK_STATE_SUCCEEDED, callbackInfo.State)
506
require.Nil(col, callbackInfo.LastAttemptFailure)
507
}
508
descCbs = append(descCbs, callbackInfo.Callback)
509
}
510
protoassert.ProtoElementsMatch(col, cbs, descCbs)
511
}, 2*time.Second, 100*time.Millisecond)
512
}
513
})
514
}
515
}
516
517
func (s *CallbacksSuite) TestNexusResetWorkflowWithCallback(opts []testcore.TestOption) {
518
env := s.newTestEnv(opts...)
519
520
ctx := s.Context()
521
sdkClient := env.SdkClient()
522
523
taskQueue := &taskqueuepb.TaskQueue{Name: env.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL}
524
workflowID := env.Tv().WorkflowID()
525
526
ch := &completionHandler{
527
requestCh: make(chan *nexusrpc.CompletionRequest, 2),
528
requestCompleteCh: make(chan error, 2),
529
}
530
defer func() {
531
close(ch.requestCh)
532
close(ch.requestCompleteCh)
533
}()
534
callbackAddress := s.runNexusCompletionHTTPServer(s.T(), ch)
535
536
// A workflow that completes once it has been reset.
537
longRunningWorkflow := func(ctx workflow.Context) error {
538
return workflow.Await(ctx, func() bool {
539
info := workflow.GetInfo(ctx)
540
541
return info.OriginalRunID != info.WorkflowExecution.RunID
542
})
543
}
544
545
env.SdkWorker().RegisterWorkflowWithOptions(longRunningWorkflow, workflow.RegisterOptions{
546
Name: "longRunningWorkflow",
547
})
548
549
cbs := []*commonpb.Callback{
550
{
551
Variant: &commonpb.Callback_Nexus_{
552
Nexus: &commonpb.Callback_Nexus{
553
Url: callbackAddress + "/cb1",
554
},
555
},
556
},
557
{
558
Variant: &commonpb.Callback_Nexus_{
559
Nexus: &commonpb.Callback_Nexus{
560
Url: callbackAddress + "/cb2",
561
},
562
},
563
},
564
}
565
566
request1 := &workflowservice.StartWorkflowExecutionRequest{
567
RequestId: uuid.NewString(),
568
Namespace: env.Namespace().String(),
569
WorkflowId: workflowID,
570
WorkflowType: &commonpb.WorkflowType{Name: "longRunningWorkflow"},
571
TaskQueue: taskQueue,
572
Input: nil,
573
Identity: s.T().Name(),
574
CompletionCallbacks: []*commonpb.Callback{cbs[0]},
575
}
576
577
startResponse1, err := env.FrontendClient().StartWorkflowExecution(ctx, request1)
578
s.NoError(err)
579
580
// Get history, iterate to ensure workflow task completed event exists.
581
workflowExecution := &commonpb.WorkflowExecution{
582
WorkflowId: workflowID,
583
RunId: startResponse1.RunId,
584
}
585
s.WaitForHistoryEvents(`
586
1 WorkflowExecutionStarted
587
2 WorkflowTaskScheduled
588
3 WorkflowTaskStarted
589
4 WorkflowTaskCompleted`,
590
env.GetHistoryFunc(env.Namespace().String(), workflowExecution),
591
5*time.Second,
592
10*time.Millisecond)
593
594
// Try starting another workflow, which will have the callback attached to the previous workflow.
595
request2 := proto.Clone(request1).(*workflowservice.StartWorkflowExecutionRequest)
596
request2.RequestId = uuid.NewString()
597
request2.WorkflowIdConflictPolicy = enumspb.WORKFLOW_ID_CONFLICT_POLICY_USE_EXISTING
598
request2.OnConflictOptions = &workflowpb.OnConflictOptions{
599
AttachRequestId: true,
600
AttachCompletionCallbacks: true,
601
}
602
request2.CompletionCallbacks = []*commonpb.Callback{cbs[1]}
603
604
startResponse2, err := env.FrontendClient().StartWorkflowExecution(ctx, request2)
605
s.NoError(err)
606
s.False(startResponse2.Started)
607
s.Equal(workflowExecution.RunId, startResponse2.RunId)
608
609
// Get history, iterate to ensure workflow execution options updated event exists.
610
s.WaitForHistoryEvents(`
611
1 WorkflowExecutionStarted
612
2 WorkflowTaskScheduled
613
3 WorkflowTaskStarted
614
4 WorkflowTaskCompleted
615
5 WorkflowExecutionOptionsUpdated`,
616
env.GetHistoryFunc(env.Namespace().String(), workflowExecution),
617
5*time.Second,
618
10*time.Millisecond)
619
620
// Reset workflow must copy all callbacks even after the reset point.
621
resetWfResponse, err := sdkClient.ResetWorkflowExecution(ctx, &workflowservice.ResetWorkflowExecutionRequest{
622
Namespace: env.Namespace().String(),
623
624
WorkflowExecution: workflowExecution,
625
Reason: "TestNexusResetWorkflowWithCallback",
626
WorkflowTaskFinishEventId: 3,
627
RequestId: "test_id",
628
})
629
s.NoError(err)
630
631
// Get the description of the run that was reset and ensure that its callback is still in STANDBY state.
632
description, err := sdkClient.DescribeWorkflowExecution(ctx, workflowID, startResponse1.RunId)
633
s.NoError(err)
634
s.Equal(enumspb.WORKFLOW_EXECUTION_STATUS_TERMINATED, description.WorkflowExecutionInfo.Status)
635
636
// Should not be invoked during a reset
637
s.Len(description.Callbacks, len(cbs))
638
descCbs := make([]*commonpb.Callback, 0, len(description.Callbacks))
639
for _, callbackInfo := range description.Callbacks {
640
s.Equal(enumspb.CALLBACK_STATE_STANDBY, callbackInfo.State)
641
s.Equal(int32(0), callbackInfo.Attempt)
642
descCbs = append(descCbs, callbackInfo.Callback)
643
}
644
s.ProtoElementsMatch(cbs, descCbs)
645
646
resetWorkflowRun := sdkClient.GetWorkflow(ctx, workflowID, resetWfResponse.RunId)
647
err = resetWorkflowRun.Get(ctx, nil)
648
s.NoError(err)
649
650
for range cbs {
651
select {
652
case completion := <-ch.requestCh:
653
s.Equal(nexus.OperationStateSucceeded, completion.State)
654
case <-time.After(time.Second):
655
s.Fail("timeout waiting for callback")
656
}
657
select {
658
case ch.requestCompleteCh <- nil:
659
case <-time.After(time.Second):
660
s.Fail("timeout writing to completion channel")
661
}
662
}
663
664
await.Require(s.Context(), s.T(),
665
func(t *await.T) {
666
// Get the description of the run post-reset and ensure its callbacks are in SUCCEEDED
667
// state.
668
description, err = sdkClient.DescribeWorkflowExecution(ctx, resetWorkflowRun.GetID(), "")
669
require.NoError(t, err)
670
require.Equal(
671
t,
672
enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED,
673
description.WorkflowExecutionInfo.Status,
674
)
675
676
require.Len(t, description.Callbacks, len(cbs))
677
descCbs = make([]*commonpb.Callback, 0, len(description.Callbacks))
678
for _, callbackInfo := range description.Callbacks {
679
require.Equal(t, enumspb.CALLBACK_STATE_SUCCEEDED, callbackInfo.State)
680
descCbs = append(descCbs, callbackInfo.Callback)
681
}
682
protoassert.ProtoElementsMatch(t, cbs, descCbs)
683
},
684
2*time.Second,
685
100*time.Millisecond,
686
)
687
}
688
689
func blockingWorkflow(ctx workflow.Context) error {
690
return workflow.Await(ctx, func() bool {
691
return false
692
})
693
}
694
695
func (s *CallbacksSuite) TestNexusResetWorkflowWithCallback_ResetToNotBaseRun(opts []testcore.TestOption) {
696
env := s.newTestEnv(opts...)
697
698
/*
699
* 1. Start WF w/ no callbacks and immediately terminate
700
* 2. Start WF second time w/ a callback
701
* 3. Reset WF back to the first run
702
* 4. Verify callback is called
703
*/
704
705
ctx := s.Context()
706
707
taskQueue := &taskqueuepb.TaskQueue{Name: env.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL}
708
workflowID := env.Tv().WorkflowID()
709
710
ch := &completionHandler{
711
requestCh: make(chan *nexusrpc.CompletionRequest, 1),
712
requestCompleteCh: make(chan error, 1),
713
}
714
defer func() {
715
close(ch.requestCh)
716
close(ch.requestCompleteCh)
717
}()
718
callbackAddress := s.runNexusCompletionHTTPServer(s.T(), ch)
719
720
env.SdkWorker().RegisterWorkflow(blockingWorkflow)
721
722
// 1. Start WF w/ no callbacks and immediately terminate
723
request1 := &workflowservice.StartWorkflowExecutionRequest{
724
RequestId: uuid.NewString(),
725
Namespace: env.Namespace().String(),
726
WorkflowId: workflowID,
727
WorkflowType: &commonpb.WorkflowType{Name: "blockingWorkflow"},
728
TaskQueue: taskQueue,
729
Input: nil,
730
WorkflowRunTimeout: durationpb.New(20 * time.Second),
731
Identity: s.T().Name(),
732
}
733
734
startResponse1, err := env.FrontendClient().StartWorkflowExecution(ctx, request1)
735
s.NoError(err)
736
737
// Validate the workflow started, then terminate it
738
workflowExecution := &commonpb.WorkflowExecution{
739
WorkflowId: workflowID,
740
RunId: startResponse1.RunId,
741
}
742
s.WaitForHistoryEvents(`
743
1 WorkflowExecutionStarted
744
2 WorkflowTaskScheduled
745
3 WorkflowTaskStarted
746
4 WorkflowTaskCompleted`,
747
env.GetHistoryFunc(env.Namespace().String(), workflowExecution),
748
5*time.Second,
749
10*time.Millisecond)
750
751
_, err = env.FrontendClient().TerminateWorkflowExecution(ctx, &workflowservice.TerminateWorkflowExecutionRequest{
752
Namespace: env.Namespace().String(),
753
WorkflowExecution: workflowExecution,
754
Reason: s.T().Name(),
755
Identity: env.Tv().WorkerIdentity(),
756
})
757
s.NoError(err)
758
759
// 2. Start WF second time w/ callbacks (new run)
760
cbs := []*commonpb.Callback{
761
{Variant: &commonpb.Callback_Nexus_{Nexus: &commonpb.Callback_Nexus{Url: callbackAddress + "/cb1"}}},
762
}
763
764
request2 := proto.Clone(request1).(*workflowservice.StartWorkflowExecutionRequest)
765
request2.RequestId = uuid.NewString()
766
request2.CompletionCallbacks = cbs
767
768
_, err = env.FrontendClient().StartWorkflowExecution(ctx, request2)
769
s.NoError(err)
770
771
// 3. Reset workflow back to the first (terminated) run as base; must copy callbacks
772
_, err = env.SdkClient().ResetWorkflowExecution(ctx, &workflowservice.ResetWorkflowExecutionRequest{
773
Namespace: env.Namespace().String(),
774
WorkflowExecution: workflowExecution, // base = first (terminated) run
775
Reason: s.T().Name(),
776
WorkflowTaskFinishEventId: 4,
777
RequestId: "test_id",
778
})
779
s.NoError(err)
780
781
// 4. Wait for callback deliveries via the handler channels
782
select {
783
case completion := <-ch.requestCh:
784
s.Equal(nexus.OperationStateFailed, completion.State)
785
ch.requestCompleteCh <- nil
786
case <-ctx.Done():
787
s.FailNow("timed out waiting for callback")
788
}
789
790
// Ensure the original workflow runs to completion to avoid leaving dangling runs
791
_, err = env.FrontendClient().TerminateWorkflowExecution(ctx, &workflowservice.TerminateWorkflowExecutionRequest{
792
Namespace: env.Namespace().String(),
793
WorkflowExecution: &commonpb.WorkflowExecution{
794
WorkflowId: workflowID,
795
},
796
Reason: s.T().Name(),
797
Identity: env.Tv().WorkerIdentity(),
798
})
799
s.NoError(err)
800
}