/
workflow.go
1905 lines (1737 loc) · 80.3 KB
/
workflow.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
// Copyright (c) 2017-2020 Uber Technologies Inc.
// Portions of the Software are attributed to Copyright (c) 2020 Temporal Technologies Inc.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
// THE SOFTWARE.
package internal
import (
"errors"
"fmt"
"reflect"
"strings"
"time"
"github.com/uber-go/tally"
"go.uber.org/zap"
s "go.uber.org/cadence/.gen/go/shared"
"go.uber.org/cadence/internal/common"
"go.uber.org/cadence/internal/common/backoff"
)
var (
errDomainNotSet = errors.New("domain is not set")
errTaskListNotSet = errors.New("task list is not set")
errWorkflowIDNotSet = errors.New("workflowId is not set")
errLocalActivityParamsBadRequest = errors.New("missing local activity parameters through context, check LocalActivityOptions")
errActivityParamsBadRequest = errors.New("missing activity parameters through context, check ActivityOptions")
errWorkflowOptionBadRequest = errors.New("missing workflow options through context, check WorkflowOptions")
errSearchAttributesNotSet = errors.New("search attributes is empty")
)
type (
// Channel must be used in workflows instead of a native Go chan.
//
// Use workflow.NewChannel(ctx) to create an unbuffered Channel instance,
// workflow.NewBufferedChannel(ctx, size) to create a Channel which has a buffer,
// or workflow.GetSignalChannel(ctx, "name") to get a Channel that contains data sent to this workflow by a call to
// SignalWorkflow (e.g. on the Client, or similar methods like SignalExternalWorkflow or SignalChildWorkflow).
//
// Both NewChannel and NewBufferedChannel have "Named" constructors as well.
// These names will be visible in stack-trace queries, so they can help with debugging, but they do not otherwise
// impact behavior at all, and are not recorded anywhere (so you can change them without versioning your code).
//
// Also note that channels created by NewChannel and NewBufferedChannel do not do any serialization or
// deserialization - you will receive whatever value was sent, and non-(de)serializable values like function
// references and interfaces are fine, the same as using a normal Go channel.
//
// Signal channels, however, contain whatever bytes were sent to your workflow, and the values must be decoded into
// the output value. By default, this means that Receive(ctx, &out) will use json.Unmarshal(data, &out), but this
// can be overridden at a worker level (worker.Options) or at a context level (workflow.WithDataConverter(ctx, dc)).
//
// You are able to send values to your own signal channels, and these values will behave the same as they do in
// normal channels (i.e. they will not be (de)serialized). However, doing so is not generally recommended, as
// mixing the value types can increase the risk that you fail to read a value, causing values to be lost. See
// Receive for more details about that behavior.
Channel interface {
// Receive blocks until it receives a value, and then assigns the received value to the provided pointer.
// It returns false when the Channel is closed and all data has already been consumed from the Channel, in the
// same way as Go channel reads work, but the assignment only occurs if there was a value in the Channel.
//
// This is technically equivalent to:
// received, ok := <- aChannel:
// if ok {
// *valuePtr = received
// }
//
// But if your output values are zero values, this is equivalent to a normal channel read:
// value, ok <- aChannel
//
// valuePtr must be assignable, and will be used to assign (for in-memory data in regular channels) or decode
// (for signal channels) the data in the channel.
//
// If decoding or assigning fails:
// - an error will be logged
// - the value will be dropped from the channel
// - Receive will automatically try again
// - This will continue until a successful value is found, or the channel is emptied and it resumes blocking.
// Closed channels with no values will always succeed, but they will not change valuePtr.
//
// Go would normally prevent incorrect-type failures like this at compile time, but the same cannot be done
// here. If you need to "try" to assign to multiple things, similar to a Future you can use:
// - for signal channels, a []byte pointer. This will give you the raw data that Cadence received, and no
// decoding will be attempted, so you can try it yourself.
// - for other channels, an interface{} pointer. All values are interfaces, so this will never fail, and you
// can inspect the type with reflection or type assertions.
Receive(ctx Context, valuePtr interface{}) (ok bool)
// ReceiveAsync tries to Receive from Channel without blocking.
// If there is data available from the Channel, it assigns the data to valuePtr and returns true.
// Otherwise, it returns false immediately.
//
// This is technically equivalent to:
// select {
// case received, ok := <- aChannel:
// if ok {
// *valuePtr = received
// }
// default:
// // no value was read
// ok = false
// }
//
// But if your output values are zero values, this is equivalent to a simpler form:
// select {
// case value, ok := <- aChannel:
// default:
// // no value was read
// ok = false
// }
//
// Decoding or assigning failures are handled like Receive.
ReceiveAsync(valuePtr interface{}) (ok bool)
// ReceiveAsyncWithMoreFlag is the same as ReceiveAsync, with an extra return to indicate if there could be
// more values from the Channel in the future.
// `more` is false only when Channel is closed and the read failed (empty).
//
// This is technically equivalent to:
// select {
// case received, ok := <- aChannel:
// if ok {
// *valuePtr = received
// }
// more = ok
// default:
// // no value was read
// ok = false
// // but the read would have blocked, so the channel is not closed
// more = true
// }
//
// But if your output values are zero values, this is equivalent to a simpler form:
// select {
// case value, ok := <- aChannel:
// more = ok
// default:
// // no value was read
// ok = false
// // but the read would have blocked, so the channel is not closed
// more = true
// }
//
// Decoding or assigning failures are handled like Receive.
ReceiveAsyncWithMoreFlag(valuePtr interface{}) (ok bool, more bool)
// Send blocks until the data is sent.
//
// This is equivalent to `aChannel <- v`.
Send(ctx Context, v interface{})
// SendAsync will try to send without blocking.
// It returns true if the data was sent (i.e. there was room in the buffer, or a reader was waiting to receive
// it), otherwise it returns false.
//
// This is equivalent to:
// select {
// case aChannel <- v: ok = true
// default: ok = false
// }
SendAsync(v interface{}) (ok bool)
// Close closes the Channel, and prohibits subsequent sends.
// As with a normal Go channel that has been closed, sending to a closed channel will panic.
Close()
}
// Selector must be used in workflows instead of a native Go select statement.
//
// Use workflow.NewSelector(ctx) to create a Selector instance, and then add cases to it with its methods.
// The interface is intended to simulate Go's select statement, and any Go select can be fairly trivially rewritten
// for a Selector with effectively identical behavior.
//
// For example, normal Go code like below (which will receive values forever, until idle for an hour):
// chA := make(chan int)
// chB := make(chan int)
// counter := 0
// for {
// select {
// case x := <- chA:
// counter += i
// case y := <- chB:
// counter += i
// case <- time.After(time.Hour):
// break
// }
// }
// can be written as:
// chA := workflow.NewChannel(ctx)
// chB := workflow.NewChannel(ctx)
// counter := 0
// for {
// timedout := false
// s := workflow.NewSelector(ctx)
// s.AddReceive(chA, func(c workflow.Channel, more bool) {
// var x int
// c.Receive(ctx, &x)
// counter += i
// })
// s.AddReceive(chB, func(c workflow.Channel, more bool) {
// var y int
// c.Receive(ctx, &y)
// counter += i
// })
// s.AddFuture(workflow.NewTimer(ctx, time.Hour), func(f workflow.Future) {
// timedout = true
// })
// s.Select(ctx)
// if timedout {
// break
// }
// }
//
// You can create a new Selector as needed or mutate one and call Select multiple times, but note that:
//
// 1. AddFuture will not behave the same across both patterns. Read AddFuture for more details.
//
// 2. There is no way to remove a case from a Selector, so you must make a new Selector to "remove" them.
//
// Finally, note that Select will not return until a condition's needs are met, like a Go selector - canceling the
// Context used to construct the Selector, or the Context used to Select, will not (directly) unblock a Select call.
// Read Select for more details.
Selector interface {
// AddReceive waits until a value can be received from a channel.
// f is invoked when the channel has data or is closed.
//
// This is equivalent to `case v, ok := <- aChannel`, and `ok` will only be false when
// the channel is both closed and no data was received.
//
// When f is invoked, the data (or closed state) remains untouched in the channel, so
// you need to `c.Receive(ctx, &out)` (or `c.ReceiveAsync(&out)`) to remove and decode the value.
// Failure to do this is not an error - the value will simply remain in the channel until a future
// Receive retrieves it.
//
// The `ok` argument will match what a call to c.Receive would return (on a successful read), so it
// may be used to check for closed + empty channels without needing to try to read from the channel.
// See Channel.Receive for additional details about reading from channels.
AddReceive(c Channel, f func(c Channel, ok bool)) Selector
// AddSend waits to send a value to a channel.
// f is invoked when the value was successfully sent to the channel.
//
// This is equivalent to `case aChannel <- value`.
//
// Unlike AddReceive, the value has already been sent on the channel when f is invoked.
AddSend(c Channel, v interface{}, f func()) Selector
// AddFuture waits until a Future is ready, and then invokes f only once.
// If the Future is ready before Select is called, it is eligible to be invoked immediately.
//
// There is no direct equivalent in a native Go select statement.
// It was added because Futures are common in Cadence code, and some patterns are much simpler with it.
//
// Each call to AddFuture will invoke its f at most one time, regardless of how many times Select is called.
// This means, for a Future that is (or will be) ready:
// - Adding the Future once, then calling Select twice, will invoke the callback once with the first Select
// call, and then wait for other Selector conditions in the second Select call (or block forever if there are
// no other eligible conditions).
// - Adding the same Future twice, then calling Select twice, will invoke each callback once.
// - Adding the same Future to two different Selectors, then calling Select once on each Selector, will invoke
// each Selector's callback once.
//
// Therefore, with a Future "f" that is or will become ready, this is an infinite loop that will consume as much
// CPU as possible:
// for {
// workflow.NewSelector(ctx).AddFuture(f, func(f workflow.Future){}).Select(ctx)
// }
// While this will loop once, and then wait idle forever:
// s := workflow.NewSelector(ctx).AddFuture(f, func(f workflow.Future){})
// for {
// s.Select(ctx)
// }
AddFuture(future Future, f func(f Future)) Selector
// AddDefault adds a default branch to the selector.
// f is invoked immediately when none of the other conditions (AddReceive, AddSend, AddFuture) are met for a
// Select call.
//
// This is equivalent to a `default:` case.
//
// Note that this applies to each Select call. If you create a Selector with only one AddDefault, and then call
// Select on it twice, f will be invoked twice.
AddDefault(f func())
// Select waits for one of the added conditions to be met and invokes the callback as described above.
// If no condition is met, Select will block until one or more are available, then one callback will be invoked.
// If no condition is ever met, Select will block forever.
//
// Note that Select does not return an error, and does not stop waiting if its Context is canceled.
// This mimics a native Go select statement, which has no way to be interrupted except for its listed cases.
//
// If you wish to stop Selecting when the Context is canceled, use AddReceive with the Context's Done() channel,
// in the same way as you would use a `case <- ctx.Done():` in a Go select statement. E.g.:
// cancelled := false
// s := workflow.NewSelector(ctx)
// s.AddFuture(f, func(f workflow.Future) {}) // assume this is never ready
// s.AddReceive(ctx.Done(), func(c workflow.Channel, more bool) {
// // this will be invoked when the Context is cancelled for any reason,
// // and more will be false.
// cancelled = true
// })
// s.Select(ctx)
// if cancelled {
// // this will be executed
// }
Select(ctx Context)
}
// WaitGroup must be used instead of native go sync.WaitGroup by
// workflow code. Use workflow.NewWaitGroup(ctx) method to create
// a new WaitGroup instance
WaitGroup interface {
Add(delta int)
Done()
Wait(ctx Context)
}
// Future represents the result of an asynchronous computation.
Future interface {
// Get blocks until the future is ready.
// When ready it either returns the Future's contained error, or assigns the contained value to the output var.
// Failures to assign or decode the value will panic.
//
// Two common patterns to retrieve data are:
// var out string
// // this will assign the string value, which may be "", or an error and leave out as "".
// err := f.Get(ctx, &out)
// and
// var out *string
// // this will assign the string value, which may be "" or nil, or an error and leave out as nil.
// err := f.Get(ctx, &out)
//
// The valuePtr parameter can be nil when the encoded result value is not needed:
// err := f.Get(ctx, nil)
//
// Futures with values set in-memory via a call to their Settable's methods can be retrieved without knowing the
// type with an interface, i.e. this will not ever panic:
// var out interface{}
// // this will assign the same value that was set,
// // and you can check its type with reflection or type assertions.
// err := f.Get(ctx, &out)
//
// Futures with encoded data from e.g. activities or child workflows can bypass decoding with a byte slice, and
// similarly this will not ever panic:
// var out []byte
// // out will contain the raw bytes given to Cadence's servers, you should decode it however is necessary
// err := f.Get(ctx, &out) // err can only be the Future's contained error
Get(ctx Context, valuePtr interface{}) error
// IsReady will return true Get is guaranteed to not block.
IsReady() bool
}
// Settable is used to set value or error on a future.
// See more: workflow.NewFuture(ctx).
Settable interface {
Set(value interface{}, err error)
SetValue(value interface{})
SetError(err error)
Chain(future Future) // Value (or error) of the future become the same of the chained one.
}
// ChildWorkflowFuture represents the result of a child workflow execution
ChildWorkflowFuture interface {
Future
// GetChildWorkflowExecution returns a future that will be ready when child workflow execution started. You can
// get the WorkflowExecution of the child workflow from the future. Then you can use Workflow ID and RunID of
// child workflow to cancel or send signal to child workflow.
// childWorkflowFuture := workflow.ExecuteChildWorkflow(ctx, child, ...)
// var childWE WorkflowExecution
// if err := childWorkflowFuture.GetChildWorkflowExecution().Get(ctx, &childWE); err == nil {
// // child workflow started, you can use childWE to get the WorkflowID and RunID of child workflow
// }
GetChildWorkflowExecution() Future
// SignalWorkflowByID sends a signal to the child workflow. This call will block until child workflow is started.
SignalChildWorkflow(ctx Context, signalName string, data interface{}) Future
}
// WorkflowType identifies a workflow type.
WorkflowType struct {
Name string
}
// WorkflowExecution Details.
WorkflowExecution struct {
ID string
RunID string
}
// EncodedValue is type alias used to encapsulate/extract encoded result from workflow/activity.
EncodedValue struct {
value []byte
dataConverter DataConverter
}
// Version represents a change version. See GetVersion call.
Version int
// ChildWorkflowOptions stores all child workflow specific parameters that will be stored inside of a Context.
// The current timeout resolution implementation is in seconds and uses math.Ceil(d.Seconds()) as the duration. But is
// subjected to change in the future.
ChildWorkflowOptions struct {
// Domain of the child workflow.
// Optional: the current workflow (parent)'s domain will be used if this is not provided.
Domain string
// WorkflowID of the child workflow to be scheduled.
// Optional: an auto generated workflowID will be used if this is not provided.
WorkflowID string
// TaskList that the child workflow needs to be scheduled on.
// Optional: the parent workflow task list will be used if this is not provided.
TaskList string
// ExecutionStartToCloseTimeout - The end to end timeout for the child workflow execution.
// Mandatory: no default
ExecutionStartToCloseTimeout time.Duration
// TaskStartToCloseTimeout - The decision task timeout for the child workflow.
// Optional: default is 10s if this is not provided (or if 0 is provided).
TaskStartToCloseTimeout time.Duration
// WaitForCancellation - Whether to wait for cancelled child workflow to be ended (child workflow can be ended
// as: completed/failed/timedout/terminated/canceled)
// Optional: default false
WaitForCancellation bool
// WorkflowIDReusePolicy - Whether server allow reuse of workflow ID, can be useful
// for dedup logic if set to WorkflowIdReusePolicyRejectDuplicate
WorkflowIDReusePolicy WorkflowIDReusePolicy
// RetryPolicy specify how to retry child workflow if error happens.
// Optional: default is no retry
RetryPolicy *RetryPolicy
// CronSchedule - Optional cron schedule for workflow. If a cron schedule is specified, the workflow will run
// as a cron based on the schedule. The scheduling will be based on UTC time. Schedule for next run only happen
// after the current run is completed/failed/timeout. If a RetryPolicy is also supplied, and the workflow failed
// or timeout, the workflow will be retried based on the retry policy. While the workflow is retrying, it won't
// schedule its next run. If next schedule is due while workflow is running (or retrying), then it will skip that
// schedule. Cron workflow will not stop until it is terminated or cancelled (by returning cadence.CanceledError).
// The cron spec is as following:
// ┌───────────── minute (0 - 59)
// │ ┌───────────── hour (0 - 23)
// │ │ ┌───────────── day of the month (1 - 31)
// │ │ │ ┌───────────── month (1 - 12)
// │ │ │ │ ┌───────────── day of the week (0 - 6) (Sunday to Saturday)
// │ │ │ │ │
// │ │ │ │ │
// * * * * *
CronSchedule string
// Memo - Optional non-indexed info that will be shown in list workflow.
Memo map[string]interface{}
// SearchAttributes - Optional indexed info that can be used in query of List/Scan/Count workflow APIs (only
// supported when Cadence server is using ElasticSearch). The key and value type must be registered on Cadence server side.
// Use GetSearchAttributes API to get valid key and corresponding value type.
SearchAttributes map[string]interface{}
// ParentClosePolicy - Optional policy to decide what to do for the child.
// Default is Terminate (if onboarded to this feature)
ParentClosePolicy ParentClosePolicy
// Bugports allows opt-in enabling of older, possibly buggy behavior, primarily intended to allow temporarily
// emulating old behavior until a fix is deployed.
//
// Bugports are always deprecated and may be removed in future versions.
// Generally speaking they will *likely* remain in place for one minor version, and then they may be removed to
// allow cleaning up the additional code complexity that they cause.
//
// Deprecated: All bugports are always deprecated and may be removed at any time.
Bugports Bugports
}
// Bugports allows opt-in enabling of older, possibly buggy behavior, primarily intended to allow temporarily
// emulating old behavior until a fix is deployed.
// By default, bugs (especially rarely-occurring ones) are fixed and all users are opted into the new behavior.
// Back-ported buggy behavior *may* be available via these flags.
//
// Fields in here are NOT guaranteed to be stable. They will almost certainly be removed in the next major
// release, and might be removed earlier if a need arises, e.g. if the historical behavior causes too much of an
// increase in code complexity.
//
// See each individual field for details.
//
// Bugports are always deprecated and may be removed in future versions.
// Generally speaking they will *likely* remain in place for one minor version, and then they may be removed to
// allow cleaning up the additional code complexity that they cause.
//
// DEPRECATED: All bugports are always deprecated and may be removed at any time.
Bugports struct {
// StartChildWorkflowsOnCanceledContext allows emulating older, buggy behavior that existed prior to v0.18.4.
//
// Prior to the fix, child workflows would be started and keep running when their context was canceled in two
// situations:
// 1) when the context was canceled before ExecuteChildWorkflow is called, and
// 2) when the context was canceled after ExecuteChildWorkflow but before the child workflow was started.
//
// 1 is unfortunately easy to trigger, though many workflows will encounter an error earlier and not reach the
// child-workflow-executing code. 2 is expected to be very rare in practice.
//
// To permanently emulate old behavior, use a disconnected context when starting child workflows, and
// cancel it only after `childfuture.GetWorkflowExecution().Get(...)` returns. This can be used when this flag
// is removed in the future.
//
// If you have currently-broken workflows and need to repair them, there are two primary options:
//
// 1: Check the BinaryChecksum value of your new deploy and/or of the decision that is currently failing
// workflows. Then set this flag when replaying history on those not-fixed checksums. Concretely, this means
// checking both `workflow.GetInfo(ctx).BinaryChecksum` (note that sufficiently old clients may not have
// recorded a value, and it may be nil) and `workflow.IsReplaying(ctx)`.
//
// 2: Reset broken workflows back to either before the buggy behavior was recorded, or before the fixed behavior
// was deployed. A "bad binary" reset type can do the latter in bulk, see the CLI's
// `cadence workflow reset-batch --reset_type BadBinary --help` for details. For the former, check the failing
// histories, identify the point at which the bug occurred, and reset to prior to that decision task.
//
// Added in 0.18.4, this may be removed in or after v0.19.0, so please migrate off of it ASAP.
//
// Deprecated: All bugports are always deprecated and may be removed at any time.
StartChildWorkflowsOnCanceledContext bool
}
)
// RegisterWorkflowOptions consists of options for registering a workflow
type RegisterWorkflowOptions struct {
Name string
// Workflow type name is equal to function name instead of fully qualified name including function package.
// This option has no effect when explicit Name is provided.
EnableShortName bool
DisableAlreadyRegisteredCheck bool
}
// RegisterWorkflow - registers a workflow function with the framework.
// The public form is: workflow.Register(...)
// A workflow takes a cadence context and input and returns a (result, error) or just error.
// Examples:
//
// func sampleWorkflow(ctx workflow.Context, input []byte) (result []byte, err error)
// func sampleWorkflow(ctx workflow.Context, arg1 int, arg2 string) (result []byte, err error)
// func sampleWorkflow(ctx workflow.Context) (result []byte, err error)
// func sampleWorkflow(ctx workflow.Context, arg1 int) (result string, err error)
//
// Serialization of all primitive types, structures is supported ... except channels, functions, variadic, unsafe pointer.
// This method calls panic if workflowFunc doesn't comply with the expected format.
// Deprecated: Global workflow registration methods are replaced by equivalent Worker instance methods.
// This method is kept to maintain backward compatibility and should not be used.
func RegisterWorkflow(workflowFunc interface{}) {
RegisterWorkflowWithOptions(workflowFunc, RegisterWorkflowOptions{})
}
// RegisterWorkflowWithOptions registers the workflow function with options.
// The public form is: workflow.RegisterWithOptions(...)
// The user can use options to provide an external name for the workflow or leave it empty if no
// external name is required. This can be used as
//
// workflow.RegisterWithOptions(sampleWorkflow, RegisterWorkflowOptions{})
// workflow.RegisterWithOptions(sampleWorkflow, RegisterWorkflowOptions{Name: "foo"})
//
// A workflow takes a cadence context and input and returns a (result, error) or just error.
// Examples:
//
// func sampleWorkflow(ctx workflow.Context, input []byte) (result []byte, err error)
// func sampleWorkflow(ctx workflow.Context, arg1 int, arg2 string) (result []byte, err error)
// func sampleWorkflow(ctx workflow.Context) (result []byte, err error)
// func sampleWorkflow(ctx workflow.Context, arg1 int) (result string, err error)
//
// Serialization of all primitive types, structures is supported ... except channels, functions, variadic, unsafe pointer.
// This method calls panic if workflowFunc doesn't comply with the expected format or tries to register the same workflow
// type name twice. Use workflow.RegisterOptions.DisableAlreadyRegisteredCheck to allow multiple registrations.
// Deprecated: Global workflow registration methods are replaced by equivalent Worker instance methods.
// This method is kept to maintain backward compatibility and should not be used.
func RegisterWorkflowWithOptions(workflowFunc interface{}, opts RegisterWorkflowOptions) {
registry := getGlobalRegistry()
registry.RegisterWorkflowWithOptions(workflowFunc, opts)
}
// GetRegisteredWorkflowTypes returns the registered workflow function/alias names.
// The public form is: workflow.GetRegisteredWorkflowTypes(...)
func GetRegisteredWorkflowTypes() []string {
registry := getGlobalRegistry()
return registry.GetRegisteredWorkflowTypes()
}
// Await blocks the calling thread until condition() returns true
// Returns CanceledError if the ctx is canceled.
func Await(ctx Context, condition func() bool) error {
state := getState(ctx)
defer state.unblocked()
for !condition() {
doneCh := ctx.Done()
// TODO: Consider always returning a channel
if doneCh != nil {
if _, more := doneCh.ReceiveAsyncWithMoreFlag(nil); !more {
return NewCanceledError("Await context cancelled")
}
}
state.yield("Await")
}
return nil
}
// NewChannel create new Channel instance
func NewChannel(ctx Context) Channel {
state := getState(ctx)
state.dispatcher.channelSequence++
return NewNamedChannel(ctx, fmt.Sprintf("chan-%v", state.dispatcher.channelSequence))
}
// NewNamedChannel create new Channel instance with a given human readable name.
// Name appears in stack traces that are blocked on this channel.
func NewNamedChannel(ctx Context, name string) Channel {
env := getWorkflowEnvironment(ctx)
return &channelImpl{name: name, dataConverter: getDataConverterFromWorkflowContext(ctx), env: env}
}
// NewBufferedChannel create new buffered Channel instance
func NewBufferedChannel(ctx Context, size int) Channel {
env := getWorkflowEnvironment(ctx)
return &channelImpl{size: size, dataConverter: getDataConverterFromWorkflowContext(ctx), env: env}
}
// NewNamedBufferedChannel create new BufferedChannel instance with a given human readable name.
// Name appears in stack traces that are blocked on this Channel.
func NewNamedBufferedChannel(ctx Context, name string, size int) Channel {
env := getWorkflowEnvironment(ctx)
return &channelImpl{name: name, size: size, dataConverter: getDataConverterFromWorkflowContext(ctx), env: env}
}
// NewSelector creates a new Selector instance.
func NewSelector(ctx Context) Selector {
state := getState(ctx)
state.dispatcher.selectorSequence++
return NewNamedSelector(ctx, fmt.Sprintf("selector-%v", state.dispatcher.selectorSequence))
}
// NewNamedSelector creates a new Selector instance with a given human readable name.
// Name appears in stack traces that are blocked on this Selector.
func NewNamedSelector(ctx Context, name string) Selector {
return &selectorImpl{name: name}
}
// NewWaitGroup creates a new WaitGroup instance.
func NewWaitGroup(ctx Context) WaitGroup {
f, s := NewFuture(ctx)
return &waitGroupImpl{future: f, settable: s}
}
// Go creates a new coroutine. It has similar semantic to goroutine in a context of the workflow.
func Go(ctx Context, f func(ctx Context)) {
state := getState(ctx)
state.dispatcher.newCoroutine(ctx, f)
}
// GoNamed creates a new coroutine with a given human readable name.
// It has similar semantic to goroutine in a context of the workflow.
// Name appears in stack traces that are blocked on this Channel.
func GoNamed(ctx Context, name string, f func(ctx Context)) {
state := getState(ctx)
state.dispatcher.newNamedCoroutine(ctx, name, f)
}
// NewFuture creates a new future as well as associated Settable that is used to set its value.
func NewFuture(ctx Context) (Future, Settable) {
impl := &futureImpl{channel: NewChannel(ctx).(*channelImpl)}
return impl, impl
}
func (wc *workflowEnvironmentInterceptor) ExecuteWorkflow(ctx Context, workflowType string, inputArgs ...interface{}) (results []interface{}) {
args := []reflect.Value{reflect.ValueOf(ctx)}
for _, arg := range inputArgs {
// []byte arguments are not serialized
switch arg.(type) {
case []byte:
args = append(args, reflect.ValueOf(arg))
default:
args = append(args, reflect.ValueOf(arg).Elem())
}
}
fnValue := reflect.ValueOf(wc.fn)
retValues := fnValue.Call(args)
for _, r := range retValues {
results = append(results, r.Interface())
}
return
}
// ExecuteActivity requests activity execution in the context of a workflow.
// Context can be used to pass the settings for this activity.
// For example: task list that this need to be routed, timeouts that need to be configured.
// Use ActivityOptions to pass down the options.
//
// ao := ActivityOptions{
// TaskList: "exampleTaskList",
// ScheduleToStartTimeout: 10 * time.Second,
// StartToCloseTimeout: 5 * time.Second,
// ScheduleToCloseTimeout: 10 * time.Second,
// HeartbeatTimeout: 0,
// }
// ctx := WithActivityOptions(ctx, ao)
//
// Or to override a single option
//
// ctx := WithTaskList(ctx, "exampleTaskList")
//
// Input activity is either an activity name (string) or a function representing an activity that is getting scheduled.
// Input args are the arguments that need to be passed to the scheduled activity.
//
// If the activity failed to complete then the future get error would indicate the failure, and it can be one of
// CustomError, TimeoutError, CanceledError, PanicError, GenericError.
// You can cancel the pending activity using context(workflow.WithCancel(ctx)) and that will fail the activity with
// error CanceledError.
//
// ExecuteActivity returns Future with activity result or failure.
func ExecuteActivity(ctx Context, activity interface{}, args ...interface{}) Future {
i := getWorkflowInterceptor(ctx)
registry := getRegistryFromWorkflowContext(ctx)
activityType := getActivityFunctionName(registry, activity)
return i.ExecuteActivity(ctx, activityType, args...)
}
func (wc *workflowEnvironmentInterceptor) ExecuteActivity(ctx Context, typeName string, args ...interface{}) Future {
// Validate type and its arguments.
dataConverter := getDataConverterFromWorkflowContext(ctx)
registry := getRegistryFromWorkflowContext(ctx)
future, settable := newDecodeFuture(ctx, typeName)
activityType, err := getValidatedActivityFunction(typeName, args, registry)
if err != nil {
settable.Set(nil, err)
return future
}
// Validate context options.
options, err := getValidatedActivityOptions(ctx)
if err != nil {
settable.Set(nil, err)
return future
}
// Validate session state.
if sessionInfo := getSessionInfo(ctx); sessionInfo != nil {
isCreationActivity := isSessionCreationActivity(typeName)
if sessionInfo.sessionState == sessionStateFailed && !isCreationActivity {
settable.Set(nil, ErrSessionFailed)
return future
}
if sessionInfo.sessionState == sessionStateOpen && !isCreationActivity {
// Use session tasklist
oldTaskListName := options.TaskListName
options.TaskListName = sessionInfo.tasklist
defer func() {
options.TaskListName = oldTaskListName
}()
}
}
// Retrieve headers from context to pass them on
header := getHeadersFromContext(ctx)
input, err := encodeArgs(dataConverter, args)
if err != nil {
panic(err)
}
params := executeActivityParams{
activityOptions: *options,
ActivityType: *activityType,
Input: input,
DataConverter: dataConverter,
Header: header,
}
ctxDone, cancellable := ctx.Done().(*channelImpl)
cancellationCallback := &receiveCallback{}
a := getWorkflowEnvironment(ctx).ExecuteActivity(params, func(r []byte, e error) {
settable.Set(r, e)
if cancellable {
// future is done, we don't need the cancellation callback anymore.
ctxDone.removeReceiveCallback(cancellationCallback)
}
})
if cancellable {
cancellationCallback.fn = func(v interface{}, more bool) bool {
if ctx.Err() == ErrCanceled {
wc.env.RequestCancelActivity(a.activityID)
}
return false
}
_, ok, more := ctxDone.receiveAsyncImpl(cancellationCallback)
if ok || !more {
cancellationCallback.fn(nil, more)
}
}
return future
}
// ExecuteLocalActivity requests to run a local activity. A local activity is like a regular activity with some key
// differences:
// * Local activity is scheduled and run by the workflow worker locally.
// * Local activity does not need Cadence server to schedule activity task and does not rely on activity worker.
// * No need to register local activity.
// * The parameter activity to ExecuteLocalActivity() must be a function.
// * Local activity is for short living activities (usually finishes within seconds).
// * Local activity cannot heartbeat.
//
// Context can be used to pass the settings for this local activity.
// For now there is only one setting for timeout to be set:
//
// lao := LocalActivityOptions{
// ScheduleToCloseTimeout: 5 * time.Second,
// }
// ctx := WithLocalActivityOptions(ctx, lao)
//
// The timeout here should be relative shorter than the DecisionTaskStartToCloseTimeout of the workflow. If you need a
// longer timeout, you probably should not use local activity and instead should use regular activity. Local activity is
// designed to be used for short living activities (usually finishes within seconds).
//
// Input args are the arguments that will to be passed to the local activity. The input args will be hand over directly
// to local activity function without serialization/deserialization because we don't need to pass the input across process
// boundary. However, the result will still go through serialization/deserialization because we need to record the result
// as history to cadence server so if the workflow crashes, a different worker can replay the history without running
// the local activity again.
//
// If the activity failed to complete then the future get error would indicate the failure, and it can be one of
// CustomError, TimeoutError, CanceledError, PanicError, GenericError.
// You can cancel the pending activity by cancel the context(workflow.WithCancel(ctx)) and that will fail the activity
// with error CanceledError.
//
// ExecuteLocalActivity returns Future with local activity result or failure.
func ExecuteLocalActivity(ctx Context, activity interface{}, args ...interface{}) Future {
i := getWorkflowInterceptor(ctx)
env := getWorkflowEnvironment(ctx)
activityType := getActivityFunctionName(env.GetRegistry(), activity)
ctx = WithValue(ctx, localActivityFnContextKey, activity)
return i.ExecuteLocalActivity(ctx, activityType, args...)
}
func (wc *workflowEnvironmentInterceptor) ExecuteLocalActivity(ctx Context, activityType string, args ...interface{}) Future {
header := getHeadersFromContext(ctx)
activityFn := ctx.Value(localActivityFnContextKey)
if activityFn == nil {
panic("ExecuteLocalActivity: Expected context key " + localActivityFnContextKey + " is missing")
}
future, settable := newDecodeFuture(ctx, activityFn)
if err := validateFunctionArgs(activityFn, args, false); err != nil {
settable.Set(nil, err)
return future
}
options, err := getValidatedLocalActivityOptions(ctx)
if err != nil {
settable.Set(nil, err)
return future
}
params := &executeLocalActivityParams{
localActivityOptions: *options,
ActivityFn: activityFn,
ActivityType: activityType,
InputArgs: args,
WorkflowInfo: GetWorkflowInfo(ctx),
DataConverter: getDataConverterFromWorkflowContext(ctx),
ScheduledTime: Now(ctx), // initial scheduled time
Header: header,
}
Go(ctx, func(ctx Context) {
for {
f := wc.scheduleLocalActivity(ctx, params)
var result []byte
err := f.Get(ctx, &result)
if retryErr, ok := err.(*needRetryError); ok && retryErr.Backoff > 0 {
// Backoff for retry
Sleep(ctx, retryErr.Backoff)
// increase the attempt, and retry the local activity
params.Attempt = retryErr.Attempt + 1
continue
}
// not more retry, return whatever is received.
settable.Set(result, err)
return
}
})
return future
}
type needRetryError struct {
Backoff time.Duration
Attempt int32
}
func (e *needRetryError) Error() string {
return fmt.Sprintf("Retry backoff: %v, Attempt: %v", e.Backoff, e.Attempt)
}
func (wc *workflowEnvironmentInterceptor) scheduleLocalActivity(ctx Context, params *executeLocalActivityParams) Future {
f := &futureImpl{channel: NewChannel(ctx).(*channelImpl)}
ctxDone, cancellable := ctx.Done().(*channelImpl)
cancellationCallback := &receiveCallback{}
la := wc.env.ExecuteLocalActivity(*params, func(lar *localActivityResultWrapper) {
if cancellable {
// future is done, we don't need cancellation anymore
ctxDone.removeReceiveCallback(cancellationCallback)
}
if lar.err == nil || IsCanceledError(lar.err) || lar.backoff <= 0 {
f.Set(lar.result, lar.err)
return
}
// set retry error, and it will be handled by workflow.ExecuteLocalActivity().
f.Set(nil, &needRetryError{Backoff: lar.backoff, Attempt: lar.attempt})
return
})
if cancellable {
cancellationCallback.fn = func(v interface{}, more bool) bool {
if ctx.Err() == ErrCanceled {
getWorkflowEnvironment(ctx).RequestCancelLocalActivity(la.activityID)
}
return false
}
_, ok, more := ctxDone.receiveAsyncImpl(cancellationCallback)
if ok || !more {
cancellationCallback.fn(nil, more)
}
}
return f
}
// ExecuteChildWorkflow requests child workflow execution in the context of a workflow.
// Context can be used to pass the settings for the child workflow.
// For example: task list that this child workflow should be routed, timeouts that need to be configured.
// Use ChildWorkflowOptions to pass down the options.
//
// cwo := ChildWorkflowOptions{
// ExecutionStartToCloseTimeout: 10 * time.Minute,
// TaskStartToCloseTimeout: time.Minute,
// }
// ctx := WithChildWorkflowOptions(ctx, cwo)
//
// Input childWorkflow is either a workflow name or a workflow function that is getting scheduled.
// Input args are the arguments that need to be passed to the child workflow function represented by childWorkflow.
// If the child workflow failed to complete then the future get error would indicate the failure and it can be one of
// CustomError, TimeoutError, CanceledError, GenericError.
// You can cancel the pending child workflow using context(workflow.WithCancel(ctx)) and that will fail the workflow with
// error CanceledError.
// ExecuteChildWorkflow returns ChildWorkflowFuture.
func ExecuteChildWorkflow(ctx Context, childWorkflow interface{}, args ...interface{}) ChildWorkflowFuture {
i := getWorkflowInterceptor(ctx)
env := getWorkflowEnvironment(ctx)
workflowType := getWorkflowFunctionName(env.GetRegistry(), childWorkflow)
return i.ExecuteChildWorkflow(ctx, workflowType, args...)
}
func (wc *workflowEnvironmentInterceptor) ExecuteChildWorkflow(ctx Context, childWorkflowType string, args ...interface{}) ChildWorkflowFuture {
mainFuture, mainSettable := newDecodeFuture(ctx, childWorkflowType)
executionFuture, executionSettable := NewFuture(ctx)
result := &childWorkflowFutureImpl{
decodeFutureImpl: mainFuture.(*decodeFutureImpl),
executionFuture: executionFuture.(*futureImpl),
}
// clients prior to v0.18.4 would incorrectly start child workflows that were started with cancelled contexts,
// and did not react to cancellation between requested and started.
correctChildCancellation := true
workflowOptionsFromCtx := getWorkflowEnvOptions(ctx)
// Starting with a canceled context should immediately fail, no need to even try.
if ctx.Err() != nil {
if workflowOptionsFromCtx.bugports.StartChildWorkflowsOnCanceledContext {
// backport the bug
correctChildCancellation = false
} else {
mainSettable.SetError(ctx.Err())
executionSettable.SetError(ctx.Err())
return result
}
}
dc := workflowOptionsFromCtx.dataConverter
env := getWorkflowEnvironment(ctx)
wfType, input, err := getValidatedWorkflowFunction(childWorkflowType, args, dc, env.GetRegistry())
if err != nil {
executionSettable.Set(nil, err)