go.temporal.io/server/tests/links_test.go
929 LOC · 0 covered · 929 uncovered · 0 ranges · 0 concepts · 0 introducers · 0 tests
1
package tests
2
3
import (
4
"context"
5
"testing"
6
"time"
7
8
"github.com/google/uuid"
9
commandpb "go.temporal.io/api/command/v1"
10
commonpb "go.temporal.io/api/common/v1"
11
enumspb "go.temporal.io/api/enums/v1"
12
taskqueuepb "go.temporal.io/api/taskqueue/v1"
13
workflowpb "go.temporal.io/api/workflow/v1"
14
"go.temporal.io/api/workflowservice/v1"
15
"go.temporal.io/api/workflowservice/v1/workflowservicenexus"
16
"go.temporal.io/sdk/client"
17
"go.temporal.io/server/common/dynamicconfig"
18
commonnexus "go.temporal.io/server/common/nexus"
19
sdkconverter "go.temporal.io/server/common/sdk"
20
"go.temporal.io/server/common/testing/parallelsuite"
21
"go.temporal.io/server/common/testing/protorequire"
22
"go.temporal.io/server/tests/testcore"
23
)
24
25
type LinksSuite struct {
26
parallelsuite.Suite[*LinksSuite]
27
}
28
29
func TestLinksTestSuite(t *testing.T) {
30
parallelsuite.Run(t, &LinksSuite{})
31
}
32
33
var links = []*commonpb.Link{
34
{
35
Variant: &commonpb.Link_WorkflowEvent_{
36
WorkflowEvent: &commonpb.Link_WorkflowEvent{
37
Namespace: "dont-care",
38
WorkflowId: "whatever",
39
RunId: uuid.NewString(),
40
},
41
},
42
},
43
}
44
45
func enableSignalBacklinkOpts() []testcore.TestOption {
46
return []testcore.TestOption{
47
testcore.WithDynamicConfig(dynamicconfig.EnableChasm, true),
48
testcore.WithDynamicConfig(dynamicconfig.EnableCHASMSignalBacklinks, true),
49
testcore.WithDynamicConfig(dynamicconfig.EnableSignalWithStartFromWorkflow, true),
50
}
51
}
52
53
// getWorkflowRunRequestInfo calls DescribeWorkflowExecution and returns the ExtendedInfo's RequestIDInfo
54
// corresponding to the supplied request ID. Fails the current test if there are any errors or if there is
55
// no RequestIDInfo corresponding to the supplied RequestID found.
56
func (s *LinksSuite) getWorkflowRunRequestInfo(
57
ctx context.Context, env *testcore.TestEnv,
58
workflowEx *commonpb.WorkflowExecution, wantRequestID string) *workflowpb.RequestIdInfo {
59
s.T().Helper()
60
61
descResp, err := env.FrontendClient().DescribeWorkflowExecution(ctx, &workflowservice.DescribeWorkflowExecutionRequest{
62
Namespace: env.Namespace().String(),
63
Execution: workflowEx,
64
})
65
s.NoError(err, "error describing workflow")
66
67
requestIDInfos := descResp.GetWorkflowExtendedInfo().GetRequestIdInfos()
68
s.Contains(requestIDInfos, wantRequestID, "No request with ID %s found in workflow execution requestIdInfos", wantRequestID)
69
70
return requestIDInfos[wantRequestID]
71
}
72
73
func (s *LinksSuite) TestTerminateWorkflow_LinksAttachedToEvent() {
74
env := testcore.NewEnv(s.T())
75
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
76
defer cancel()
77
run, err := env.SdkClient().ExecuteWorkflow(
78
ctx,
79
client.StartWorkflowOptions{
80
TaskQueue: "dont-care",
81
},
82
"test-workflow-type",
83
)
84
s.NoError(err)
85
86
_, err = env.FrontendClient().TerminateWorkflowExecution(ctx, &workflowservice.TerminateWorkflowExecutionRequest{
87
Namespace: env.Namespace().String(),
88
WorkflowExecution: &commonpb.WorkflowExecution{
89
WorkflowId: run.GetID(),
90
},
91
Reason: "test",
92
Links: links,
93
})
94
s.NoError(err)
95
96
// TODO(bergundy): Use SdkClient if and when it exposes links on TerminateWorkflow.
97
history := env.SdkClient().GetWorkflowHistory(ctx, run.GetID(), "", false, enumspb.HISTORY_EVENT_FILTER_TYPE_CLOSE_EVENT)
98
event, err := history.Next()
99
s.NoError(err)
100
protorequire.ProtoSliceEqual(s.T(), links, event.Links)
101
}
102
103
func (s *LinksSuite) TestRequestCancelWorkflow_LinksAttachedToEvent() {
104
env := testcore.NewEnv(s.T())
105
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
106
defer cancel()
107
run, err := env.SdkClient().ExecuteWorkflow(
108
ctx,
109
client.StartWorkflowOptions{
110
TaskQueue: "dont-care",
111
},
112
"test-workflow-type",
113
)
114
s.NoError(err)
115
116
_, err = env.FrontendClient().RequestCancelWorkflowExecution(ctx, &workflowservice.RequestCancelWorkflowExecutionRequest{
117
Namespace: env.Namespace().String(),
118
WorkflowExecution: &commonpb.WorkflowExecution{
119
WorkflowId: run.GetID(),
120
},
121
Reason: "test",
122
Links: links,
123
})
124
s.NoError(err)
125
126
// TODO(bergundy): Use SdkClient if and when it exposes links on CancelWorkflow.
127
history := env.SdkClient().GetWorkflowHistory(ctx, run.GetID(), "", false, enumspb.HISTORY_EVENT_FILTER_TYPE_ALL_EVENT)
128
foundEvent := false
129
for history.HasNext() {
130
event, err := history.Next()
131
s.NoError(err)
132
if event.EventType != enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_CANCEL_REQUESTED {
133
continue
134
}
135
foundEvent = true
136
protorequire.ProtoSliceEqual(s.T(), links, event.Links)
137
}
138
s.True(foundEvent)
139
}
140
141
func (s *LinksSuite) TestSignalWorkflowExecution_LinksAttachedToEvent() {
142
env := testcore.NewEnv(s.T(), enableSignalBacklinkOpts()...)
143
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
144
defer cancel()
145
146
signalTest := newSignalWorkflowTest(env, s)
147
148
// Start the workflow.
149
startResult := signalTest.startTargetWorkflow(ctx)
150
targetWorkflowID, targetWorkflowRunID := startResult.WorkflowID, startResult.WorkflowRunID
151
152
// Signal the workflow (for the first time).
153
signalWorkflowRequestID := uuid.NewString()
154
signalResp := signalTest.signalWorkflow(ctx, targetWorkflowID, signalWorkflowRequestID)
155
gotLink := signalResp.GetLink()
156
157
wantLink := &commonpb.Link{
158
Variant: &commonpb.Link_WorkflowEvent_{
159
WorkflowEvent: &commonpb.Link_WorkflowEvent{
160
Namespace: env.Namespace().String(),
161
WorkflowId: targetWorkflowID,
162
RunId: targetWorkflowRunID,
163
Reference: &commonpb.Link_WorkflowEvent_RequestIdRef{
164
RequestIdRef: &commonpb.Link_WorkflowEvent_RequestIdReference{
165
RequestId: signalWorkflowRequestID,
166
EventType: enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED,
167
},
168
},
169
},
170
},
171
}
172
protorequire.ProtoEqual(s.T(), wantLink, gotLink)
173
174
// Second call with same RequestId hits the dedup path but must still return the same link.
175
signalResp2 := signalTest.signalWorkflow(ctx, targetWorkflowID, signalWorkflowRequestID)
176
protorequire.ProtoEqual(s.T(), wantLink, signalResp2.GetLink())
177
178
// Confirm no duplicate events in the Workflow's history.
179
history := env.SdkClient().GetWorkflowHistory(ctx, targetWorkflowID, "", false, enumspb.HISTORY_EVENT_FILTER_TYPE_ALL_EVENT)
180
foundEvent := false
181
foundDuplicatedEvent := false
182
var signaledEventID int64
183
for history.HasNext() {
184
event, err := history.Next()
185
s.NoError(err)
186
if event.EventType != enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED {
187
continue
188
}
189
if foundEvent {
190
foundDuplicatedEvent = true
191
} else {
192
signaledEventID = event.GetEventId()
193
}
194
foundEvent = true
195
protorequire.ProtoSliceEqual(s.T(), links, event.Links)
196
}
197
s.True(foundEvent)
198
s.False(foundDuplicatedEvent, "second signal with same RequestId should be deduped and not produce a second event")
199
200
// Verify the requestID is tracked and resolves to the correct event ID.
201
workflowEx := &commonpb.WorkflowExecution{
202
WorkflowId: targetWorkflowID,
203
}
204
gotRequestInfo := s.getWorkflowRunRequestInfo(ctx, env, workflowEx, signalWorkflowRequestID)
205
s.Equal(enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED, gotRequestInfo.GetEventType())
206
s.Equal(signaledEventID, gotRequestInfo.GetEventId(), "requestID map entry must point to the SIGNALED event in history")
207
}
208
209
func (s *LinksSuite) TestSignalWithStartWorkflowExecution_LinksAttachedToEvent() {
210
// Body of the test. We run it twice, where the workflow targeted by SignalWithStart
211
// is or is-not running.
212
testImpl := func(ls *LinksSuite, signalExistingWorkflow bool) {
213
env := testcore.NewEnv(ls.T(), enableSignalBacklinkOpts()...)
214
ctx := ls.Context()
215
216
signalTest := newSignalWorkflowTest(env, ls)
217
218
// Potentially start the workflow.
219
targetWorkflowID := uuid.NewString()
220
if signalExistingWorkflow {
221
signalTest.startTargetWorkflowWithWorkflowID(ctx, targetWorkflowID)
222
}
223
224
// Send a signal to the new or existing workflow, get its RunID.
225
signalResp := signalTest.signalWithStartWorkflow(ctx, targetWorkflowID)
226
if signalExistingWorkflow {
227
ls.False(signalResp.Started)
228
} else {
229
ls.True(signalResp.Started)
230
}
231
232
targetWorkflowRunID := signalResp.GetRunId()
233
gotLink := signalResp.GetSignalLink()
234
ls.NotNil(gotLink, "no SignalLink in response")
235
236
// We don't know the RequestID for the StartWithSignal call until after it is made.
237
signalWorkflowRequestID := gotLink.GetWorkflowEvent().GetRequestIdRef().GetRequestId()
238
ls.NotEmpty(signalWorkflowRequestID, "didn't get RequestID from SignalWithStart response")
239
240
wantLink := &commonpb.Link{
241
Variant: &commonpb.Link_WorkflowEvent_{
242
WorkflowEvent: &commonpb.Link_WorkflowEvent{
243
Namespace: env.Namespace().String(),
244
WorkflowId: targetWorkflowID,
245
RunId: targetWorkflowRunID,
246
Reference: &commonpb.Link_WorkflowEvent_RequestIdRef{
247
RequestIdRef: &commonpb.Link_WorkflowEvent_RequestIdReference{
248
RequestId: signalWorkflowRequestID,
249
EventType: enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED,
250
},
251
},
252
},
253
},
254
}
255
protorequire.ProtoEqual(ls.T(), wantLink, gotLink)
256
257
// NOTE: Unlike the SignalWorkflow- version of this test, we do not verify
258
// any dedupe paths, because calling SignalWithStart twice via the Nexus
259
// endpoint will result in sending two signals to the workflow with no way
260
// to make the request idempotent. (Which is expected and by-design.)
261
262
// Verify the requestID is tracked and resolves to the correct event ID.
263
workflowEx := &commonpb.WorkflowExecution{
264
WorkflowId: targetWorkflowID,
265
}
266
gotRequestInfo := ls.getWorkflowRunRequestInfo(ctx, env, workflowEx, signalWorkflowRequestID)
267
ls.Equal(enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED, gotRequestInfo.GetEventType())
268
ls.Positive(gotRequestInfo.GetEventId())
269
}
270
271
s.Run("SignalExistingWorkflow", func(ls *LinksSuite) {
272
testImpl(ls, true)
273
})
274
s.Run("SignalStartsNewWorkflow", func(ls *LinksSuite) {
275
testImpl(ls, false)
276
})
277
}
278
279
// TestSignalWorkflowExecution_BacklinkSurvivesReset verifies that after a workflow is reset,
280
// the new run's CHASM IncomingSignals map is rebuilt from history so that DescribeWorkflow
281
// continues to return a valid requestID -> event-ID backlink for signals that occurred before
282
// the reset point.
283
//
284
// This exercises the rebuild/replay path through ApplyWorkflowExecutionSignaled, which uses
285
// the event's real event ID (not common.BufferedEventID) when writing to the CHASM tree.
286
func (s *LinksSuite) TestSignalWorkflowExecution_BacklinkSurvivesReset() {
287
env := testcore.NewEnv(s.T(), enableSignalBacklinkOpts()...)
288
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
289
defer cancel()
290
291
signalTest := newSignalWorkflowTest(env, s)
292
293
// Start the workflow.
294
startResult := signalTest.startTargetWorkflow(ctx)
295
targetWorkflowID, targetWorkflowRunID := startResult.WorkflowID, startResult.WorkflowRunID
296
297
// Signal the workflow. The signal will be included in the first WFT batch, so it will
298
// appear in history before the WFT completion event.
299
signalRequestID := uuid.NewString()
300
signalTest.signalWorkflow(ctx, targetWorkflowID, signalRequestID)
301
302
// Poll and complete the WFT so the signal is flushed to history with a real event ID.
303
pollResp, pollErr := env.FrontendClient().PollWorkflowTaskQueue(ctx, &workflowservice.PollWorkflowTaskQueueRequest{
304
Namespace: env.Namespace().String(),
305
TaskQueue: &taskqueuepb.TaskQueue{Name: signalTest.taskQueueName, Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
306
Identity: "test",
307
})
308
s.NoError(pollErr)
309
s.NotNil(pollResp.GetTaskToken())
310
_, completeErr := env.FrontendClient().RespondWorkflowTaskCompleted(ctx, &workflowservice.RespondWorkflowTaskCompletedRequest{
311
Namespace: env.Namespace().String(),
312
Identity: "test",
313
TaskToken: pollResp.TaskToken,
314
})
315
s.NoError(completeErr)
316
317
// Find the WFT completed event ID in the original run's history.
318
var wftCompletedEventID int64
319
history := env.SdkClient().GetWorkflowHistory(ctx, targetWorkflowID, targetWorkflowRunID, false, enumspb.HISTORY_EVENT_FILTER_TYPE_ALL_EVENT)
320
for history.HasNext() {
321
event, histErr := history.Next()
322
s.NoError(histErr)
323
if event.EventType == enumspb.EVENT_TYPE_WORKFLOW_TASK_COMPLETED {
324
wftCompletedEventID = event.EventId
325
break
326
}
327
}
328
s.Positive(wftCompletedEventID, "WFT completed event not found in history")
329
330
// Reset the workflow to the first WFT completion. The signal event is before this point,
331
// so it will be included in the new run's replayed history.
332
resetResp, err := env.FrontendClient().ResetWorkflowExecution(ctx, &workflowservice.ResetWorkflowExecutionRequest{
333
Namespace: env.Namespace().String(),
334
WorkflowExecution: &commonpb.WorkflowExecution{
335
WorkflowId: targetWorkflowID,
336
RunId: targetWorkflowRunID,
337
},
338
Reason: "testing-backlink-survival",
339
RequestId: uuid.NewString(),
340
WorkflowTaskFinishEventId: wftCompletedEventID,
341
})
342
s.NoError(err)
343
newRunID := resetResp.RunId
344
s.NotEmpty(newRunID)
345
346
// During reset, ApplyWorkflowExecutionSignaled rebuilds the CHASM IncomingSignals map
347
// from history, so the backlink should be present once the new run is created.
348
descResp, descErr := env.FrontendClient().DescribeWorkflowExecution(ctx, &workflowservice.DescribeWorkflowExecutionRequest{
349
Namespace: env.Namespace().String(),
350
Execution: &commonpb.WorkflowExecution{WorkflowId: targetWorkflowID, RunId: newRunID},
351
})
352
s.NoError(descErr)
353
_, signalExists := descResp.GetWorkflowExtendedInfo().GetRequestIdInfos()[signalRequestID]
354
s.True(signalExists)
355
356
// Verify the backlink on the new run points to a real (non-buffered) SIGNALED event.
357
workflowEx := &commonpb.WorkflowExecution{
358
WorkflowId: targetWorkflowID,
359
RunId: newRunID,
360
}
361
gotRequestIDInfo := s.getWorkflowRunRequestInfo(ctx, env, workflowEx, signalRequestID)
362
s.Equal(enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED, gotRequestIDInfo.GetEventType())
363
s.Positive(gotRequestIDInfo.GetEventId(), "backlink event ID must be a real, non-buffered event ID in the new run's history")
364
s.False(gotRequestIDInfo.GetBuffered())
365
}
366
367
func (s *LinksSuite) TestSignalWithStartWorkflowExecution_BacklinkSurvivesReset() {
368
// Body of the test. We run it twice, where the workflow targeted by SignalWithStart
369
// is or is-not running.
370
testImpl := func(ls *LinksSuite, signalExistingWorkflow bool) {
371
env := testcore.NewEnv(ls.T(), enableSignalBacklinkOpts()...)
372
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
373
defer cancel()
374
375
signalTest := newSignalWorkflowTest(env, ls)
376
377
// Start the workflow depending on the test scenario.
378
targetWorkflowID := uuid.NewString()
379
var targetWorkflowRunID string
380
381
if signalExistingWorkflow {
382
startResp := signalTest.startTargetWorkflowWithWorkflowID(ctx, targetWorkflowID)
383
gotWfID, gotRunID := startResp.WorkflowID, startResp.WorkflowRunID
384
ls.Equal(targetWorkflowID, gotWfID)
385
targetWorkflowRunID = gotRunID
386
}
387
388
// Signal the workflow. The signal will be included in the first WFT batch, so it will
389
// appear in history before the WFT completion event.
390
signalWithStartResp := signalTest.signalWithStartWorkflow(ctx, targetWorkflowID)
391
if signalExistingWorkflow {
392
ls.False(signalWithStartResp.Started)
393
} else {
394
ls.True(signalWithStartResp.Started)
395
}
396
397
// We don't know the RequestID for the StartWithSignal call until after it is made.
398
ls.NotNil(signalWithStartResp.GetSignalLink())
399
signalWorkflowRequestID := signalWithStartResp.GetSignalLink().GetWorkflowEvent().GetRequestIdRef().GetRequestId()
400
ls.NotEmpty(signalWorkflowRequestID, "didn't get RequestID from SignalWithStart response")
401
402
if signalExistingWorkflow {
403
ls.False(signalWithStartResp.Started)
404
ls.Equal(targetWorkflowRunID, signalWithStartResp.RunId)
405
} else {
406
ls.True(signalWithStartResp.Started)
407
targetWorkflowRunID = signalWithStartResp.GetRunId()
408
}
409
410
// Poll and complete the WFT so the signal is flushed to history with a real event ID.
411
pollResp, pollErr := env.FrontendClient().PollWorkflowTaskQueue(ctx, &workflowservice.PollWorkflowTaskQueueRequest{
412
Namespace: env.Namespace().String(),
413
TaskQueue: &taskqueuepb.TaskQueue{Name: signalTest.taskQueueName, Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
414
Identity: "test",
415
})
416
ls.NoError(pollErr)
417
ls.NotNil(pollResp.GetTaskToken())
418
_, completeErr := env.FrontendClient().RespondWorkflowTaskCompleted(ctx, &workflowservice.RespondWorkflowTaskCompletedRequest{
419
Namespace: env.Namespace().String(),
420
Identity: "test",
421
TaskToken: pollResp.TaskToken,
422
})
423
ls.NoError(completeErr)
424
425
// Find the WFT completed event ID in the original run's history.
426
var wftCompletedEventID int64
427
history := env.SdkClient().GetWorkflowHistory(ctx, targetWorkflowID, targetWorkflowRunID, false, enumspb.HISTORY_EVENT_FILTER_TYPE_ALL_EVENT)
428
for history.HasNext() {
429
event, histErr := history.Next()
430
ls.NoError(histErr)
431
if event.EventType == enumspb.EVENT_TYPE_WORKFLOW_TASK_COMPLETED {
432
wftCompletedEventID = event.EventId
433
break
434
}
435
}
436
ls.Positive(wftCompletedEventID, "WFT completed event not found in history")
437
438
// Reset the workflow to the first WFT completion. The signal event is before this point,
439
// so it will be included in the new run's replayed history.
440
resetResp, err := env.FrontendClient().ResetWorkflowExecution(ctx, &workflowservice.ResetWorkflowExecutionRequest{
441
Namespace: env.Namespace().String(),
442
WorkflowExecution: &commonpb.WorkflowExecution{
443
WorkflowId: targetWorkflowID,
444
RunId: targetWorkflowRunID,
445
},
446
Reason: "testing-backlink-survival",
447
RequestId: uuid.NewString(),
448
WorkflowTaskFinishEventId: wftCompletedEventID,
449
})
450
ls.NoError(err)
451
newRunID := resetResp.RunId
452
ls.NotEmpty(newRunID)
453
454
// Confirm the original signal is in the original workflow run's RequestID map.
455
originalWorkflowEx := &commonpb.WorkflowExecution{
456
WorkflowId: targetWorkflowID,
457
RunId: targetWorkflowRunID,
458
}
459
origSignalInfo := ls.getWorkflowRunRequestInfo(ctx, env, originalWorkflowEx, signalWorkflowRequestID)
460
ls.Equal(enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED, origSignalInfo.GetEventType())
461
ls.False(origSignalInfo.GetBuffered())
462
463
// During reset, ApplyWorkflowExecutionSignaled rebuilds the CHASM IncomingSignals map
464
// from history, so the backlink should be present once the new run is created.
465
// Verify the backlink on the new run points to a real (non-buffered) SIGNALED event.
466
resetWorkflowEx := &commonpb.WorkflowExecution{
467
WorkflowId: targetWorkflowID,
468
RunId: newRunID,
469
}
470
resetSignalInfo := ls.getWorkflowRunRequestInfo(ctx, env, resetWorkflowEx, signalWorkflowRequestID)
471
ls.Equal(enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED, resetSignalInfo.GetEventType())
472
ls.Positive(resetSignalInfo.GetEventId(), "backlink event ID must be a real, non-buffered event ID in the new run's history")
473
ls.False(resetSignalInfo.GetBuffered())
474
}
475
476
s.Run("SignalExistingWorkflow", func(ls *LinksSuite) {
477
testImpl(ls, true)
478
})
479
s.Run("SignalStartsNewWorkflow", func(ls *LinksSuite) {
480
testImpl(ls, false)
481
})
482
}
483
484
// TestSignalWorkflowExecution_BufferedDuringWorkflowTask verifies that when a signal arrives
485
// while a workflow task is being processed, DescribeWorkflow reports the backlink as buffered.
486
// Once the workflow task completes and the signal is flushed to history, the backlink must
487
// reflect a real (non-buffered) event ID.
488
func (s *LinksSuite) TestSignalWorkflowExecution_BufferedDuringWorkflowTask() {
489
env := testcore.NewEnv(s.T(), enableSignalBacklinkOpts()...)
490
ctx := s.Context()
491
492
signalTest := newSignalWorkflowTest(env, s)
493
494
// Start the workflow.
495
startResult := signalTest.startTargetWorkflow(ctx)
496
targetWorkflowID, targetWorkflowRunID := startResult.WorkflowID, startResult.WorkflowRunID
497
498
// Poll to move the WFT into "started" state to have the server wait for us to complete it.
499
// This will force the signal to stay in the buffer until the task is finished.
500
pollResp, err := env.FrontendClient().PollWorkflowTaskQueue(s.Context(), &workflowservice.PollWorkflowTaskQueueRequest{
501
Namespace: env.Namespace().String(),
502
TaskQueue: &taskqueuepb.TaskQueue{Name: signalTest.taskQueueName, Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
503
Identity: "test",
504
})
505
s.NoError(err)
506
s.NotNil(pollResp.GetTaskToken())
507
508
// This signal will be buffered since there is a WFT in-flight.
509
signalRequestID := uuid.NewString()
510
signalTest.signalWorkflow(ctx, targetWorkflowID, signalRequestID)
511
512
// WFT is still running: backlink must be present and marked buffered.
513
workflowEx := &commonpb.WorkflowExecution{
514
WorkflowId: targetWorkflowID,
515
RunId: targetWorkflowRunID,
516
}
517
gotRequestInfo := s.getWorkflowRunRequestInfo(ctx, env, workflowEx, signalRequestID)
518
s.Equal(enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED, gotRequestInfo.GetEventType())
519
s.True(gotRequestInfo.GetBuffered(), "backlink must be buffered while WFT is in progress")
520
521
// Complete the WFT, which flushes the signal to DB with a concrete EventID.
522
_, err = env.FrontendClient().RespondWorkflowTaskCompleted(s.Context(), &workflowservice.RespondWorkflowTaskCompletedRequest{
523
Namespace: env.Namespace().String(),
524
Identity: "test",
525
TaskToken: pollResp.TaskToken,
526
})
527
s.NoError(err)
528
529
// After WFT completion the backlink must resolve to a real, non-buffered event.
530
gotRequestInfo2 := s.getWorkflowRunRequestInfo(ctx, env, workflowEx, signalRequestID)
531
s.Equal(enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED, gotRequestInfo2.GetEventType())
532
s.False(gotRequestInfo2.GetBuffered(), "backlink must not be buffered after WFT completion")
533
s.Positive(gotRequestInfo2.GetEventId(), "backlink must reference a real event ID after WFT completion")
534
}
535
536
func (s *LinksSuite) TestSignalWithStartWorkflowExecution_BufferedDuringWorkflowTask() {
537
// Body of the test. We run it twice, where the workflow targeted by SignalWithStart
538
// is or is-not running.
539
testImpl := func(ls *LinksSuite, signalExistingWorkflow bool) {
540
env := testcore.NewEnv(ls.T(), enableSignalBacklinkOpts()...)
541
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
542
defer cancel()
543
544
signalTest := newSignalWorkflowTest(env, ls)
545
546
// Potentially start the workflow.
547
targetWorkflowID := uuid.NewString()
548
if signalExistingWorkflow {
549
signalTest.startTargetWorkflowWithWorkflowID(ctx, targetWorkflowID)
550
}
551
552
// Poll to move the WFT into "started" state to have the server wait for us to complete it.
553
// This will force the signal to stay in the buffer until the task is finished.
554
//
555
// We skip this step if there is no existing workflow to target with SignalWithStart, because
556
// that would have the Poll call hang until the deadline is hit. (Because there is no workflow
557
// with tasks to be executed.)
558
var pollTaskToken []byte
559
if signalExistingWorkflow {
560
pollResp, err := env.FrontendClient().PollWorkflowTaskQueue(ctx, &workflowservice.PollWorkflowTaskQueueRequest{
561
Namespace: env.Namespace().String(),
562
TaskQueue: &taskqueuepb.TaskQueue{Name: signalTest.taskQueueName, Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
563
Identity: "test",
564
})
565
ls.NoError(err)
566
ls.NotNil(pollResp.GetTaskToken())
567
568
pollTaskToken = pollResp.GetTaskToken()
569
}
570
571
// Call SignalWithStart. This will result in the event getting buffered (if the workflow
572
// is already running), or simply starting as new workflow execution.
573
signalWithStartResp := signalTest.signalWithStartWorkflow(ctx, targetWorkflowID)
574
if signalExistingWorkflow {
575
ls.False(signalWithStartResp.Started)
576
} else {
577
ls.True(signalWithStartResp.Started)
578
}
579
580
ls.NotNil(signalWithStartResp.GetSignalLink())
581
targetWorkflowRunID := signalWithStartResp.GetRunId()
582
583
signalWorkflowRequestID := signalWithStartResp.GetSignalLink().GetWorkflowEvent().GetRequestIdRef().GetRequestId()
584
ls.NotEmpty(signalWorkflowRequestID, "didn't get RequestID from SignalWithStart response")
585
586
// Get the RequestIDInfos for the running workflow.
587
workflowEx := &commonpb.WorkflowExecution{
588
WorkflowId: targetWorkflowID,
589
RunId: targetWorkflowRunID,
590
}
591
gotRequestInfo := ls.getWorkflowRunRequestInfo(ctx, env, workflowEx, signalWorkflowRequestID)
592
ls.Equal(enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED, gotRequestInfo.GetEventType())
593
594
if signalExistingWorkflow {
595
// If the signal was sent to an existing workflow, we expect the new event to be buffered.
596
ls.True(gotRequestInfo.GetBuffered(), "backlink must be buffered while WFT is in progress")
597
598
// Complete the WFT, which flushes the signal to DB with a concrete EventID.
599
_, err := env.FrontendClient().RespondWorkflowTaskCompleted(ctx, &workflowservice.RespondWorkflowTaskCompletedRequest{
600
Namespace: env.Namespace().String(),
601
Identity: "test",
602
TaskToken: pollTaskToken,
603
})
604
ls.NoError(err)
605
606
// After WFT completion the backlink must resolve to a real, non-buffered event.
607
gotRequestInfo2 := ls.getWorkflowRunRequestInfo(ctx, env, workflowEx, signalWorkflowRequestID)
608
ls.Equal(enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED, gotRequestInfo2.GetEventType())
609
ls.False(gotRequestInfo2.GetBuffered(), "backlink must not be buffered after WFT completion")
610
ls.Positive(gotRequestInfo2.GetEventId(), "backlink must reference a real event ID after WFT completion")
611
} else {
612
// If the call to SignalWithStart triggered spinning up a new workflow execution, then no buffering is necessary.
613
// We don't need to complete the WTF, because there isn't a WFT that we were polling on.
614
ls.False(gotRequestInfo.GetBuffered(), "did not expect event to be buffered")
615
ls.Positive(gotRequestInfo.GetEventId())
616
}
617
}
618
619
s.Run("SignalExistingWorkflow", func(ls *LinksSuite) {
620
testImpl(ls, true)
621
})
622
s.Run("SignalStartsNewWorkflow", func(ls *LinksSuite) {
623
testImpl(ls, false)
624
})
625
}
626
627
func (s *LinksSuite) TestSignalWithStartWorkflowExecution_LinksAttachedToRelevantEvents() {
628
env := testcore.NewEnv(s.T(), enableSignalBacklinkOpts()...)
629
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
630
defer cancel()
631
632
workflowID := testcore.RandomizeStr(s.T().Name())
633
634
request := &workflowservice.SignalWithStartWorkflowExecutionRequest{
635
Namespace: env.Namespace().String(),
636
WorkflowId: workflowID,
637
WorkflowType: &commonpb.WorkflowType{
638
Name: "dont-care",
639
},
640
SignalName: "dont-care",
641
Identity: "test",
642
TaskQueue: &taskqueuepb.TaskQueue{
643
Name: "dont-care",
644
},
645
RequestId: uuid.NewString(),
646
Links: links,
647
}
648
649
// TODO(bergundy): Use SdkClient if and when it exposes links on SignalWithStartWorkflow.
650
resp, err := env.FrontendClient().SignalWithStartWorkflowExecution(ctx, request)
651
s.NoError(err)
652
firstRunID := resp.GetRunId()
653
protorequire.ProtoEqual(
654
s.T(),
655
&commonpb.Link{
656
Variant: &commonpb.Link_WorkflowEvent_{
657
WorkflowEvent: &commonpb.Link_WorkflowEvent{
658
Namespace: env.Namespace().String(),
659
WorkflowId: workflowID,
660
RunId: firstRunID,
661
Reference: &commonpb.Link_WorkflowEvent_RequestIdRef{
662
RequestIdRef: &commonpb.Link_WorkflowEvent_RequestIdReference{
663
RequestId: request.RequestId,
664
EventType: enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED,
665
},
666
},
667
},
668
},
669
},
670
resp.GetSignalLink(),
671
)
672
673
firstRequestID := request.RequestId
674
675
// Send a second request and verify that the new signal has links attached to it too.
676
request.RequestId = uuid.NewString()
677
resp, err = env.FrontendClient().SignalWithStartWorkflowExecution(ctx, request)
678
s.NoError(err)
679
// Expect backlinks with the same RunID as before since the workflow execution didn't change,
680
// but the signal requestID should differ since this is a different request.
681
protorequire.ProtoEqual(
682
s.T(),
683
&commonpb.Link{
684
Variant: &commonpb.Link_WorkflowEvent_{
685
WorkflowEvent: &commonpb.Link_WorkflowEvent{
686
Namespace: env.Namespace().String(),
687
WorkflowId: workflowID,
688
RunId: resp.GetRunId(),
689
Reference: &commonpb.Link_WorkflowEvent_RequestIdRef{
690
RequestIdRef: &commonpb.Link_WorkflowEvent_RequestIdReference{
691
RequestId: request.RequestId, // This requestID should differ from the first backlink.
692
EventType: enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED,
693
},
694
},
695
},
696
},
697
},
698
resp.GetSignalLink(),
699
)
700
701
history := env.SdkClient().GetWorkflowHistory(ctx, workflowID, "", false, enumspb.HISTORY_EVENT_FILTER_TYPE_ALL_EVENT)
702
foundStartEvent := false
703
foundFirstSignal := false
704
foundSecondSignal := false
705
var firstSignalEventID, secondSignalEventID int64
706
for history.HasNext() {
707
event, err := history.Next()
708
s.NoError(err)
709
if event.EventType == enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED {
710
if foundFirstSignal {
711
foundSecondSignal = true
712
secondSignalEventID = event.GetEventId()
713
} else {
714
foundFirstSignal = true
715
firstSignalEventID = event.GetEventId()
716
}
717
protorequire.ProtoSliceEqual(s.T(), links, event.Links)
718
}
719
if event.EventType == enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED {
720
foundStartEvent = true
721
protorequire.ProtoSliceEqual(s.T(), links, event.Links)
722
}
723
}
724
s.True(foundStartEvent)
725
s.True(foundFirstSignal)
726
s.True(foundSecondSignal)
727
728
// Verify both requestIDs are tracked and resolve to the correct signal event IDs.
729
descResp, err := env.FrontendClient().DescribeWorkflowExecution(ctx, &workflowservice.DescribeWorkflowExecutionRequest{
730
Namespace: env.Namespace().String(),
731
Execution: &commonpb.WorkflowExecution{
732
WorkflowId: workflowID,
733
},
734
})
735
s.NoError(err)
736
requestIDInfos := descResp.GetWorkflowExtendedInfo().GetRequestIdInfos()
737
738
s.Contains(requestIDInfos, firstRequestID)
739
firstInfo := requestIDInfos[firstRequestID]
740
s.Equal(enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED, firstInfo.GetEventType())
741
s.Equal(firstSignalEventID, firstInfo.GetEventId(), "first requestID map entry must point to the first SIGNALED event in history")
742
743
s.Contains(requestIDInfos, request.RequestId)
744
secondInfo := requestIDInfos[request.RequestId]
745
s.Equal(enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED, secondInfo.GetEventType())
746
s.Equal(secondSignalEventID, secondInfo.GetEventId(), "second requestID map entry must point to the second SIGNALED event in history")
747
}
748
749
// signalWorkflowTest provides common operations for starting and signaling workflows.
750
type signalWorkflowTest struct {
751
taskQueueName string
752
workflowName string
753
754
s *LinksSuite
755
env *testcore.TestEnv
756
}
757
758
func newSignalWorkflowTest(env *testcore.TestEnv, s *LinksSuite) *signalWorkflowTest {
759
return &signalWorkflowTest{
760
taskQueueName: "test-task-queue",
761
workflowName: "test-workflow",
762
s: s,
763
env: env,
764
}
765
}
766
767
type startTargetWorkflowOutput struct {
768
WorkflowID string
769
WorkflowRunID string
770
}
771
772
// startTargetWorkflow starts a generic workflow.
773
func (swt *signalWorkflowTest) startTargetWorkflow(ctx context.Context) startTargetWorkflowOutput {
774
swt.s.T().Helper()
775
// By not supplying a WorkflowID, it will default to UUID.
776
return swt.startTargetWorkflowWithWorkflowID(ctx, "")
777
}
778
779
// startTargetWorkflowWithWorkflowID starts a workflow using the supplied Workflow ID.
780
func (swt *signalWorkflowTest) startTargetWorkflowWithWorkflowID(ctx context.Context, workflowID string) startTargetWorkflowOutput {
781
swt.s.T().Helper()
782
run, err := swt.env.SdkClient().ExecuteWorkflow(
783
ctx,
784
client.StartWorkflowOptions{
785
ID: workflowID,
786
TaskQueue: swt.taskQueueName,
787
},
788
"test-workflow-type",
789
)
790
swt.s.NoError(err)
791
792
return startTargetWorkflowOutput{
793
WorkflowID: run.GetID(),
794
WorkflowRunID: run.GetRunID(),
795
}
796
}
797
798
func (swt *signalWorkflowTest) signalWorkflow(ctx context.Context, targetWorkflowID, requestID string) *workflowservice.SignalWorkflowExecutionResponse {
799
swt.s.T().Helper()
800
req := &workflowservice.SignalWorkflowExecutionRequest{
801
Namespace: swt.env.Namespace().String(),
802
WorkflowExecution: &commonpb.WorkflowExecution{
803
WorkflowId: targetWorkflowID,
804
// Target the latest execution of the workflow.
805
RunId: "",
806
},
807
SignalName: "dont care",
808
Identity: "test",
809
RequestId: requestID,
810
Links: links,
811
}
812
// TODO(bergundy): Use SdkClient if and when it exposes links on SignalWorkflow.
813
resp, err := swt.env.FrontendClient().SignalWorkflowExecution(ctx, req)
814
swt.s.NoError(err)
815
816
return resp
817
}
818
819
// signalWithStartWorkflow invokes the SignalWithStart handler using the System Nexus Endpoint,
820
// and NOT the typical frontend API directly. This is a newer codepath that wraps the History
821
// service's API within the Nexus machinery.
822
//
823
// IMPORTANT: The RequestID CANNOT be supplied when using the Nexus variant of the SignalWithStart
824
// call, because the RequestID will be set when when doing the execution.
825
func (swt *signalWorkflowTest) signalWithStartWorkflow(ctx context.Context, targetWorkflowID string) *workflowservice.SignalWithStartWorkflowExecutionResponse {
826
swt.s.T().Helper()
827
startWithSignalRequest := &workflowservice.SignalWithStartWorkflowExecutionRequest{
828
Namespace: swt.env.Namespace().String(),
829
WorkflowId: targetWorkflowID,
830
WorkflowType: &commonpb.WorkflowType{Name: swt.workflowName},
831
TaskQueue: &taskqueuepb.TaskQueue{Name: swt.taskQueueName, Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
832
SignalName: "dont care",
833
Identity: "test",
834
835
// QUIRK: When using the system Nexus endpoint, we cannot suppply new links.
836
Links: nil,
837
// QUIRK: Must be left empty when making the call via Nexus.
838
RequestId: "",
839
}
840
841
// HERE BE DRAGONS
842
//
843
// To call the SignalWithStartWorkflowExecution found within the system Nexus endpoint, we start
844
// and wait on a trivial workflow which will actually make the call.
845
//
846
// ... HOWEVER, the go.temporal.io/[email protected] (and earlier) panics in workflow.NewNexusClient when
847
// the endpoint name starts with the reserved "__temporal_" prefix. (Which is the case when trying
848
// to target the system Nexus endpoint.)
849
//
850
// So instead, we work around this by following after signal_with_start_from_workflow_test.go's
851
// TestBothWorkflowsVisibleAfterSWSFromWorkflowProtoBinary, which drives the workflow task manually
852
// sending the command.COMMAND_TYPE_SCHEDULE_NEXUS_OPERATION.
853
//
854
// We can remove this when the SDK allows creating a Nexus client which can target the system handler.
855
const nexusCallerTaskQueue = "totally-different-task-queue"
856
857
// Start a caller workflow to obtain an initial workflow task.
858
callerRun, err := swt.env.SdkClient().ExecuteWorkflow(ctx, client.StartWorkflowOptions{
859
TaskQueue: nexusCallerTaskQueue,
860
}, "caller-workflow")
861
swt.s.NoError(err)
862
defer func() {
863
_ = swt.env.SdkClient().TerminateWorkflow(ctx, callerRun.GetID(), callerRun.GetRunID(), "test cleanup")
864
}()
865
866
// Encode the SWS request as binary/protobuf. PreferProtoDataConverter places
867
// ProtoPayloadConverter first, so proto messages are marshalled to binary/protobuf
868
// rather than the JSON proto encoding that the SDK uses by default.
869
pls, err := sdkconverter.PreferProtoDataConverter.ToPayloads(startWithSignalRequest)
870
swt.s.NoError(err)
871
swt.s.Len(pls.Payloads, 1)
872
873
protoBinaryPayload := pls.Payloads[0]
874
swt.s.Equal("binary/protobuf", string(protoBinaryPayload.Metadata["encoding"]))
875
876
// First poll: respond with a ScheduleNexusOperation command carrying the proto binary input.
877
pollResp, err := swt.env.FrontendClient().PollWorkflowTaskQueue(ctx, &workflowservice.PollWorkflowTaskQueueRequest{
878
Namespace: swt.env.Namespace().String(),
879
TaskQueue: &taskqueuepb.TaskQueue{Name: nexusCallerTaskQueue, Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
880
})
881
swt.s.NoError(err)
882
883
_, err = swt.env.FrontendClient().RespondWorkflowTaskCompleted(ctx, &workflowservice.RespondWorkflowTaskCompletedRequest{
884
TaskToken: pollResp.TaskToken,
885
Commands: []*commandpb.Command{
886
{
887
CommandType: enumspb.COMMAND_TYPE_SCHEDULE_NEXUS_OPERATION,
888
Attributes: &commandpb.Command_ScheduleNexusOperationCommandAttributes{
889
ScheduleNexusOperationCommandAttributes: &commandpb.ScheduleNexusOperationCommandAttributes{
890
Endpoint: commonnexus.SystemEndpoint,
891
Service: workflowservicenexus.TemporalAPIWorkflowserviceV1WorkflowService.ServiceName,
892
Operation: "SignalWithStartWorkflowExecution",
893
Input: protoBinaryPayload,
894
},
895
},
896
},
897
},
898
})
899
swt.s.NoError(err)
900
901
// Second poll: wait for NexusOperationCompleted or NexusOperationFailed.
902
pollResp, err = swt.env.FrontendClient().PollWorkflowTaskQueue(ctx, &workflowservice.PollWorkflowTaskQueueRequest{
903
Namespace: swt.env.Namespace().String(),
904
TaskQueue: &taskqueuepb.TaskQueue{Name: nexusCallerTaskQueue, Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
905
})
906
swt.s.NoError(err)
907
908
// Pull the StartWithSignalWorkflowExecutionResponse from the history events.
909
var (
910
startWithSignalResponse workflowservice.SignalWithStartWorkflowExecutionResponse
911
found bool
912
)
913
for _, event := range pollResp.History.Events {
914
if attrs := event.GetNexusOperationCompletedEventAttributes(); attrs != nil {
915
found = true
916
convErr := sdkconverter.PreferProtoDataConverter.FromPayloads(
917
&commonpb.Payloads{Payloads: []*commonpb.Payload{attrs.Result}},
918
&startWithSignalResponse,
919
)
920
swt.s.NoError(convErr)
921
}
922
if attrs := event.GetNexusOperationFailedEventAttributes(); attrs != nil {
923
swt.s.Fail("expected NexusOperationCompleted but got NexusOperationFailed: " + attrs.Failure.GetMessage())
924
}
925
}
926
swt.s.True(found, "did not see Nexus operation complete in workflow events")
927
928
return &startWithSignalResponse
929
}