go.temporal.io/server/components/nexusoperations/tasks.go
365 LOC · 63 covered · 302 uncovered · 28 ranges · 292 concepts · 11 introducers · 130 tests
File neighbourhood
The centred file is linked to every concept that introduces one of its ranges, every test that runs code from the file, and the gray connector concepts standing between those tests and the file's own introducer concepts. Undirected links join concepts to every file where they introduce source and concepts to the tests they introduce; arrows show specialization between the displayed concepts and bridge only concepts omitted from this view. Concept colors match the source ranges below; connector concepts have no source color and are shown in gray.
Focused file, its introducer and connector concepts, their introduced files, and tests that run code from the file
In the embedded map, ordinary wheel input scrolls the page; use the visible controls to zoom and drag to pan. Open the full-screen map for canvas navigation: wheel pans, Ctrl/Command plus wheel zooms, and arrow keys pan when this region is focused. On touch screens, open the full-screen map to pan or pinch. If JavaScript or WebGL is unavailable, use the related-file, concept, and source links on this page.
Graph controls are ready.
Interactive rendering requires JavaScript and WebGL. Use the related-file, concept, and source links on this page while the interactive map is unavailable.
1
package nexusoperations
2
3
import (
4
"errors"
5
"fmt"
6
"time"
7
8
enumspb "go.temporal.io/api/enums/v1"
9
"go.temporal.io/api/serviceerror"
10
enumsspb "go.temporal.io/server/api/enums/v1"
11
persistencespb "go.temporal.io/server/api/persistence/v1"
12
"go.temporal.io/server/common/persistence/serialization"
13
"go.temporal.io/server/service/history/consts"
14
"go.temporal.io/server/service/history/hsm"
15
"google.golang.org/protobuf/proto"
16
)
17
18
const (
19
TaskTypeInvocation = "nexusoperations.Invocation"
20
TaskTypeBackoff = "nexusoperations.Backoff"
21
TaskTypeCancelation = "nexusoperations.Cancelation"
22
TaskTypeCancelationBackoff = "nexusoperations.CancelationBackoff"
23
// NOTE: the name `Timeout` is used for backward compatibility with existing persisted tasks and predates the addition of more flexible timeout types.
24
TaskTypeScheduleToCloseTimeout = "nexusoperations.Timeout"
25
TaskTypeScheduleToStartTimeout = "nexusoperations.ScheduleToStartTimeout"
26
TaskTypeStartToCloseTimeout = "nexusoperations.StartToCloseTimeout"
27
)
28
29
var errSerializationCast = errors.New("cannot serialize HSM task. unable to cast to expected type")
30
31
type ScheduleToCloseTimeoutTask struct {
32
deadline time.Time
33
}
34
35
var _ hsm.Task = ScheduleToCloseTimeoutTask{}
36
38
>
return TaskTypeScheduleToCloseTimeout
39
>
}
40
42
>
return t.deadline
43
>
}
44
46
>
return ""
47
>
}
48
49
// Validate checks if the timeout task is still valid to execute for the given node state.
50
>
func (t ScheduleToCloseTimeoutTask) Validate(ref *persistencespb.StateMachineRef, node *hsm.Node) error {
tasks.go ×12
51
>
if err := node.CheckRunning(); err != nil {
52
return err
53
}
55
>
if err != nil {
56
return err
57
}
59
return fmt.Errorf(
60
"%w: %w: cannot timeout machine in state %v",
61
consts.ErrStaleReference,
62
hsm.ErrInvalidTransition,
63
op.State(),
64
)
65
}
67
}
68
69
type TimeoutTaskSerializer struct{}
70
71
func (TimeoutTaskSerializer) Deserialize(data []byte, attrs hsm.TaskAttributes) (hsm.Task, error) {
72
return ScheduleToCloseTimeoutTask{deadline: attrs.Deadline}, nil
73
}
74
76
>
return nil, nil
77
>
}
78
79
type InvocationTask struct {
80
EndpointName string
81
Attempt int32
82
}
83
84
var _ hsm.Task = InvocationTask{}
85
87
>
return TaskTypeInvocation
88
>
}
89
91
>
return hsm.Immediate
92
>
}
93
95
>
return t.EndpointName
96
>
}
97
98
>
func (InvocationTask) Validate(ref *persistencespb.StateMachineRef, node *hsm.Node) error {
tasks.go ×12
99
>
if err := node.CheckRunning(); err != nil {
100
return err
101
}
102
>
return hsm.ValidateState[enumsspb.NexusOperationState, Operation](node, enumsspb.NEXUS_OPERATION_STATE_SCHEDULED)
tasks.go ×12
103
}
104
105
type InvocationTaskSerializer struct{}
106
107
func (InvocationTaskSerializer) Deserialize(data []byte, attrs hsm.TaskAttributes) (hsm.Task, error) {
108
var info persistencespb.NexusInvocationTaskInfo
109
err := proto.Unmarshal(data, &info)
110
if err != nil {
111
return nil, serialization.NewDeserializationError(enumspb.ENCODING_TYPE_PROTO3, err)
112
}
113
return InvocationTask{EndpointName: attrs.Destination, Attempt: info.Attempt}, nil
114
}
115
117
>
switch task := task.(type) {
118
>
case InvocationTask:
119
>
return proto.Marshal(&persistencespb.NexusInvocationTaskInfo{Attempt: task.Attempt})
120
default:
121
return nil, serviceerror.NewInternalf("unknown HSM task type while serializing: %v", task)
122
}
123
}
124
125
type BackoffTask struct {
126
deadline time.Time
127
}
128
129
var _ hsm.Task = BackoffTask{}
130
132
>
return TaskTypeBackoff
133
>
}
134
136
>
return t.deadline
137
>
}
138
139
func (t BackoffTask) Destination() string {
140
return ""
141
}
142
143
func (t BackoffTask) Validate(_ *persistencespb.StateMachineRef, node *hsm.Node) error {
144
if err := node.CheckRunning(); err != nil {
145
return err
146
}
147
return hsm.ValidateState[enumsspb.NexusOperationState, Operation](node, enumsspb.NEXUS_OPERATION_STATE_BACKING_OFF)
148
}
149
150
type BackoffTaskSerializer struct{}
151
152
func (BackoffTaskSerializer) Deserialize(data []byte, attrs hsm.TaskAttributes) (hsm.Task, error) {
153
return BackoffTask{deadline: attrs.Deadline}, nil
154
}
155
156
func (BackoffTaskSerializer) Serialize(hsm.Task) ([]byte, error) {
157
return nil, nil
158
}
159
160
type CancelationTask struct {
161
EndpointName string
162
Attempt int32
163
}
164
165
var _ hsm.Task = CancelationTask{}
166
168
>
return TaskTypeCancelation
169
>
}
170
171
func (CancelationTask) Deadline() time.Time {
172
return hsm.Immediate
173
}
174
175
func (t CancelationTask) Destination() string {
176
return t.EndpointName
177
}
178
179
func (CancelationTask) Validate(ref *persistencespb.StateMachineRef, node *hsm.Node) error {
180
if err := node.CheckRunning(); err != nil {
181
return err
182
}
183
return hsm.ValidateState[enumspb.NexusOperationCancellationState, Cancelation](node, enumspb.NEXUS_OPERATION_CANCELLATION_STATE_SCHEDULED)
184
}
185
186
type CancelationTaskSerializer struct{}
187
188
func (CancelationTaskSerializer) Deserialize(data []byte, attrs hsm.TaskAttributes) (hsm.Task, error) {
189
var info persistencespb.NexusCancelationTaskInfo
190
err := proto.Unmarshal(data, &info)
191
if err != nil {
192
return nil, serialization.NewDeserializationError(enumspb.ENCODING_TYPE_PROTO3, err)
193
}
194
return CancelationTask{EndpointName: attrs.Destination, Attempt: info.Attempt}, nil
195
}
196
197
func (CancelationTaskSerializer) Serialize(task hsm.Task) ([]byte, error) {
198
switch task := task.(type) {
199
case CancelationTask:
200
return proto.Marshal(&persistencespb.NexusCancelationTaskInfo{Attempt: task.Attempt})
201
default:
202
return nil, serviceerror.NewInternalf("unknown HSM task type while serializing: %v", task)
203
}
204
}
205
206
type CancelationBackoffTask struct {
207
deadline time.Time
208
}
209
210
var _ hsm.Task = CancelationBackoffTask{}
211
213
>
return TaskTypeCancelationBackoff
214
>
}
215
217
>
return t.deadline
218
>
}
219
220
func (CancelationBackoffTask) Destination() string {
221
return ""
222
}
223
224
func (CancelationBackoffTask) Validate(ref *persistencespb.StateMachineRef, node *hsm.Node) error {
225
if err := node.CheckRunning(); err != nil {
226
return err
227
}
228
return hsm.ValidateState[enumspb.NexusOperationCancellationState, Cancelation](node, enumspb.NEXUS_OPERATION_CANCELLATION_STATE_BACKING_OFF)
229
}
230
231
type CancelationBackoffTaskSerializer struct{}
232
233
func (CancelationBackoffTaskSerializer) Deserialize(data []byte, attrs hsm.TaskAttributes) (hsm.Task, error) {
234
return CancelationBackoffTask{deadline: attrs.Deadline}, nil
235
}
236
237
func (CancelationBackoffTaskSerializer) Serialize(hsm.Task) ([]byte, error) {
238
return nil, nil
239
}
240
241
type ScheduleToStartTimeoutTask struct {
242
deadline time.Time
243
}
244
245
var _ hsm.Task = ScheduleToStartTimeoutTask{}
246
248
>
return TaskTypeScheduleToStartTimeout
249
>
}
250
251
func (t ScheduleToStartTimeoutTask) Deadline() time.Time {
252
return t.deadline
253
}
254
255
func (ScheduleToStartTimeoutTask) Destination() string {
256
return ""
257
}
258
259
// Validate checks if the schedule-to-start timeout task is still valid.
260
// Only valid if operation is still in SCHEDULED or BACKING_OFF state.
261
func (t ScheduleToStartTimeoutTask) Validate(ref *persistencespb.StateMachineRef, node *hsm.Node) error {
262
if err := node.CheckRunning(); err != nil {
263
return err
264
}
265
op, err := hsm.MachineData[Operation](node)
266
if err != nil {
267
return err
268
}
269
// Only timeout if we haven't started yet
270
switch op.State() {
271
case enumsspb.NEXUS_OPERATION_STATE_SCHEDULED,
272
enumsspb.NEXUS_OPERATION_STATE_BACKING_OFF:
273
return nil
274
default:
275
// Already started or completed, timeout not applicable
276
return fmt.Errorf(
277
"%w: %w: cannot apply schedule-to-start timeout to machine in state %v",
278
consts.ErrStaleReference,
279
hsm.ErrInvalidTransition,
280
op.State(),
281
)
282
}
283
}
284
285
type ScheduleToStartTimeoutTaskSerializer struct{}
286
287
func (ScheduleToStartTimeoutTaskSerializer) Deserialize(data []byte, attrs hsm.TaskAttributes) (hsm.Task, error) {
288
return ScheduleToStartTimeoutTask{deadline: attrs.Deadline}, nil
289
}
290
291
func (ScheduleToStartTimeoutTaskSerializer) Serialize(hsm.Task) ([]byte, error) {
292
return nil, nil
293
}
294
295
type StartToCloseTimeoutTask struct {
296
deadline time.Time
297
}
298
299
var _ hsm.Task = StartToCloseTimeoutTask{}
300
302
>
return TaskTypeStartToCloseTimeout
303
>
}
304
305
func (t StartToCloseTimeoutTask) Deadline() time.Time {
306
return t.deadline
307
}
308
309
func (StartToCloseTimeoutTask) Destination() string {
310
return ""
311
}
312
313
// Validate checks if the start-to-close timeout task is still valid.
314
// Only valid if operation is in STARTED state.
315
func (t StartToCloseTimeoutTask) Validate(ref *persistencespb.StateMachineRef, node *hsm.Node) error {
316
if err := node.CheckRunning(); err != nil {
317
return err
318
}
319
op, err := hsm.MachineData[Operation](node)
320
if err != nil {
321
return err
322
}
323
// Only timeout if we're in started state
324
if op.State() != enumsspb.NEXUS_OPERATION_STATE_STARTED {
325
return fmt.Errorf(
326
"%w: %w: cannot apply start-to-close timeout to machine in state %v",
327
consts.ErrStaleReference,
328
hsm.ErrInvalidTransition,
329
op.State(),
330
)
331
}
332
return nil
333
}
334
335
type StartToCloseTimeoutTaskSerializer struct{}
336
337
func (StartToCloseTimeoutTaskSerializer) Deserialize(data []byte, attrs hsm.TaskAttributes) (hsm.Task, error) {
338
return StartToCloseTimeoutTask{deadline: attrs.Deadline}, nil
339
}
340
341
func (StartToCloseTimeoutTaskSerializer) Serialize(hsm.Task) ([]byte, error) {
342
return nil, nil
343
}
344
346
>
if err := reg.RegisterTaskSerializer(TaskTypeScheduleToCloseTimeout, TimeoutTaskSerializer{}); err != nil {
347
return err
348
}
349
>
if err := reg.RegisterTaskSerializer(TaskTypeInvocation, InvocationTaskSerializer{}); err != nil {
tasks.go ×7
350
return err
351
}
352
>
if err := reg.RegisterTaskSerializer(TaskTypeBackoff, BackoffTaskSerializer{}); err != nil {
tasks.go ×7
353
return err
354
}
355
>
if err := reg.RegisterTaskSerializer(TaskTypeCancelation, CancelationTaskSerializer{}); err != nil {
tasks.go ×7
356
return err
357
}
358
>
if err := reg.RegisterTaskSerializer(TaskTypeCancelationBackoff, CancelationBackoffTaskSerializer{}); err != nil { // nolint:revive
tasks.go ×7
359
return err
360
}
361
>
if err := reg.RegisterTaskSerializer(TaskTypeScheduleToStartTimeout, ScheduleToStartTimeoutTaskSerializer{}); err != nil {
tasks.go ×7
362
return err
363
}
364
>
return reg.RegisterTaskSerializer(TaskTypeStartToCloseTimeout, StartToCloseTimeoutTaskSerializer{})
tasks.go ×7
365
}