go.temporal.io/server/components/nexusoperations/statemachine.go
682 LOC · 298 covered · 384 uncovered · 97 ranges · 475 concepts · 51 introducers · 214 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
failurepb "go.temporal.io/api/failure/v1"
10
historypb "go.temporal.io/api/history/v1"
11
"go.temporal.io/api/serviceerror"
12
enumsspb "go.temporal.io/server/api/enums/v1"
13
persistencespb "go.temporal.io/server/api/persistence/v1"
14
"go.temporal.io/server/common/backoff"
15
"go.temporal.io/server/service/history/hsm"
16
"google.golang.org/protobuf/proto"
17
"google.golang.org/protobuf/types/known/timestamppb"
18
)
19
20
const (
21
// OperationMachineType is a unique type identifier for the Operation state machine.
22
OperationMachineType = "nexusoperations.Operation"
23
24
// CancelationMachineType is a unique type identifier for the Cancelation state machine.
25
CancelationMachineType = "nexusoperations.Cancelation"
26
27
// A marker for the first return value from a progress() that indicates the machine is in a terminal state.
28
// TODO: Remove this once transition history is fully implemented.
29
terminalStage = 3
30
)
31
32
// CancelationMachineKey is a fixed key for the cancelation machine as a child of the operation machine.
33
var CancelationMachineKey = hsm.Key{Type: CancelationMachineType, ID: ""}
34
35
// MachineCollection creates a new typed [statemachines.Collection] for operations.
37
>
return hsm.NewCollection[Operation](tree, OperationMachineType)
38
>
}
39
40
// Operation state machine.
41
type Operation struct {
42
*persistencespb.NexusOperationInfo
43
}
44
45
// AddChild adds a new operation child machine to the given node and transitions it to the SCHEDULED state.
46
>
func AddChild(node *hsm.Node, id string, event *historypb.HistoryEvent, eventToken []byte) (*hsm.Node, error) {
statemachine.go ×14
47
>
attrs := event.GetNexusOperationScheduledEventAttributes()
48
>
49
>
node, err := node.AddChild(hsm.Key{Type: OperationMachineType, ID: id}, Operation{
50
>
&persistencespb.NexusOperationInfo{
51
>
EndpointId: attrs.EndpointId,
52
>
Endpoint: attrs.Endpoint,
53
>
Service: attrs.Service,
54
>
Operation: attrs.Operation,
55
>
ScheduledTime: event.EventTime,
56
>
ScheduleToCloseTimeout: attrs.ScheduleToCloseTimeout,
57
>
ScheduleToStartTimeout: attrs.ScheduleToStartTimeout,
58
>
StartToCloseTimeout: attrs.StartToCloseTimeout,
59
>
RequestId: attrs.RequestId,
60
>
State: enumsspb.NEXUS_OPERATION_STATE_UNSPECIFIED,
61
>
ScheduledEventToken: eventToken,
62
>
},
63
>
})
64
>
65
>
if err != nil {
66
return nil, err
67
}
68
69
>
return node, hsm.MachineTransition(node, func(op Operation) (hsm.TransitionOutput, error) {
statemachine.go ×14
70
>
output, err := TransitionScheduled.Apply(op, EventScheduled{Node: node})
71
>
if err != nil {
72
return output, err
73
}
75
>
if err != nil {
76
return output, err
77
}
79
>
return output, err
80
})
81
}
82
84
>
return o.NexusOperationInfo.State
85
>
}
86
88
>
o.NexusOperationInfo.State = state
89
>
}
90
92
>
o.NexusOperationInfo.Attempt++
93
>
o.NexusOperationInfo.LastAttemptCompleteTime = timestamppb.New(ts)
94
>
o.NexusOperationInfo.LastAttemptFailure = nil
95
>
}
96
97
func (o Operation) cancelRequested(node *hsm.Node) (bool, error) {
98
_, err := node.Child([]hsm.Key{CancelationMachineKey})
99
if err == nil {
100
return true, nil
101
}
102
if errors.Is(err, hsm.ErrStateMachineNotFound) {
103
return false, nil
104
}
105
return false, err
106
}
107
109
>
child, err := node.Child([]hsm.Key{CancelationMachineKey})
110
>
if errors.Is(err, hsm.ErrStateMachineNotFound) {
111
return nil, nil
112
}
114
return nil, err
115
}
117
>
return &cancelation, err
118
}
119
121
>
child, err := node.Child([]hsm.Key{CancelationMachineKey})
122
>
if errors.Is(err, hsm.ErrStateMachineNotFound) {
124
>
}
126
}
127
128
// transitionTasks returns tasks that are emitted as transition outputs.
130
>
switch o.State() { // nolint:exhaustive
132
>
return []hsm.Task{BackoffTask{deadline: o.NextAttemptScheduleTime.AsTime()}}, nil
134
>
return []hsm.Task{InvocationTask{EndpointName: o.Endpoint, Attempt: o.Attempt}}, nil
136
>
return nil, nil
137
}
138
}
139
140
// creationTasks returns tasks that are emitted when the machine is created.
142
>
var tasks []hsm.Task
143
>
144
>
if o.ScheduleToCloseTimeout.AsDuration() != 0 {
146
>
deadline: o.ScheduledTime.AsTime().Add(o.ScheduleToCloseTimeout.AsDuration()),
147
>
})
148
>
}
149
152
>
deadline: o.ScheduledTime.AsTime().Add(o.ScheduleToStartTimeout.AsDuration()),
153
>
})
154
>
}
155
157
}
158
159
// startToCloseTimeoutTask returns the StartToCloseTimeout task if the timeout is set.
160
// This task is created when an operation transitions to the STARTED state.
162
>
if o.StartedTime.AsTime().IsZero() || o.StartToCloseTimeout.AsDuration() == 0 {
164
>
}
166
>
StartToCloseTimeoutTask{
167
>
deadline: o.StartedTime.AsTime().Add(o.StartToCloseTimeout.AsDuration()),
168
>
},
169
>
}
170
}
171
173
>
transitionTasks, err := o.transitionTasks()
174
>
if err != nil {
175
return nil, err
176
}
178
>
if err != nil {
179
return nil, err
180
}
181
>
return append(append(transitionTasks, creationTasks...), o.startToCloseTimeoutTask()...), nil
statemachine.go ×3
182
}
183
185
>
tasks, err := o.transitionTasks()
186
>
if err != nil {
187
return hsm.TransitionOutput{}, err
188
}
190
}
191
192
type operationMachineDefinition struct{}
193
195
>
return OperationMachineType
196
>
}
197
198
func (operationMachineDefinition) Deserialize(d []byte) (any, error) {
199
info := &persistencespb.NexusOperationInfo{}
200
return Operation{info}, proto.Unmarshal(d, info)
201
}
202
204
>
if state, ok := state.(Operation); ok {
205
>
return proto.Marshal(state.NexusOperationInfo)
206
>
}
207
return nil, fmt.Errorf("invalid operation provided: %v", state)
208
}
209
210
// CompareState compares the progress of two Operation state machines to determine whether to sync machine state while
211
// processing a replication task.
212
// TODO: Remove this implementation once transition history is fully implemented.
213
>
func (operationMachineDefinition) CompareState(state1, state2 any) (int, error) {
statemachine.go ×6
214
>
o1, ok := state1.(Operation)
215
>
if !ok {
216
return 0, fmt.Errorf("%w: expected state1 to be a Operation instance, got %v", hsm.ErrIncompatibleType, state1)
217
}
219
>
if !ok {
220
return 0, fmt.Errorf("%w: expected state2 to be a Operation instance, got %v", hsm.ErrIncompatibleType, state2)
221
}
222
224
>
if err != nil {
225
return 0, fmt.Errorf("failed to get progress for state1: %w", err)
226
}
228
>
if err != nil {
229
return 0, fmt.Errorf("failed to get progress for state2: %w", err)
230
}
233
>
}
235
>
return 0, serviceerror.NewInvalidArgumentf("cannot compare two distinct terminal states: %v, %v", o1.State(), o2.State())
nexus.pb.go ×1
236
>
}
238
}
239
240
// EventScheduled is triggered when the operation is meant to be scheduled - immediately after initialization.
241
type EventScheduled struct {
242
Node *hsm.Node
243
}
244
245
var TransitionScheduled = hsm.NewTransition(
246
[]enumsspb.NexusOperationState{enumsspb.NEXUS_OPERATION_STATE_UNSPECIFIED},
247
enumsspb.NEXUS_OPERATION_STATE_SCHEDULED,
249
>
return op.output()
250
>
},
251
)
252
253
// EventRescheduled is triggered when the operation is meant to be rescheduled after backing off from a previous
254
// attempt.
255
type EventRescheduled struct {
256
Node *hsm.Node
257
}
258
259
var TransitionRescheduled = hsm.NewTransition(
260
[]enumsspb.NexusOperationState{enumsspb.NEXUS_OPERATION_STATE_BACKING_OFF},
261
enumsspb.NEXUS_OPERATION_STATE_SCHEDULED,
263
>
op.NextAttemptScheduleTime = nil
264
>
return op.output()
265
>
},
266
)
267
268
// EventAttemptFailed is triggered when an invocation attempt is failed with a retryable error.
269
type EventAttemptFailed struct {
270
Time time.Time
271
Failure *failurepb.Failure
272
Node *hsm.Node
273
RetryPolicy backoff.RetryPolicy
274
}
275
276
var TransitionAttemptFailed = hsm.NewTransition(
277
[]enumsspb.NexusOperationState{enumsspb.NEXUS_OPERATION_STATE_SCHEDULED},
278
enumsspb.NEXUS_OPERATION_STATE_BACKING_OFF,
279
>
func(op Operation, event EventAttemptFailed) (hsm.TransitionOutput, error) {
statemachine.go ×2
280
>
op.recordAttempt(event.Time)
281
>
// Use 0 for elapsed time as we don't limit the retry by time (for now).
282
>
// The last argument (error) is ignored.
283
>
nextDelay := event.RetryPolicy.ComputeNextDelay(0, int(op.Attempt), nil)
284
>
nextAttemptScheduleTime := event.Time.Add(nextDelay)
285
>
op.NextAttemptScheduleTime = timestamppb.New(nextAttemptScheduleTime)
286
>
op.LastAttemptFailure = event.Failure
287
>
return op.output()
288
>
},
289
)
290
291
// EventFailed is triggered when an invocation attempt is failed with a non retryable error.
292
type EventFailed struct {
293
Time time.Time
294
Node *hsm.Node
295
Attributes *historypb.NexusOperationFailedEventAttributes
296
}
297
298
var TransitionFailed = hsm.NewTransition(
299
[]enumsspb.NexusOperationState{
300
enumsspb.NEXUS_OPERATION_STATE_SCHEDULED,
301
enumsspb.NEXUS_OPERATION_STATE_BACKING_OFF,
302
enumsspb.NEXUS_OPERATION_STATE_STARTED,
303
},
304
enumsspb.NEXUS_OPERATION_STATE_FAILED,
306
>
// Not recording the last attempt information here since the state machine will be deleted immediately after this transition.
307
>
// If we ever use this code for a standalone state machine implementation we will want to record the last
308
>
// attempt information in case the completion is a result of a synchronous operation.
309
>
return op.output()
310
>
},
311
)
312
313
// EventSucceeded is triggered when an invocation attempt succeeds.
314
type EventSucceeded struct {
315
// Only set if the operation completed synchronously, as a response to a StartOperation RPC.
316
Time time.Time
317
Node *hsm.Node
318
}
319
320
var TransitionSucceeded = hsm.NewTransition(
321
[]enumsspb.NexusOperationState{
322
enumsspb.NEXUS_OPERATION_STATE_SCHEDULED,
323
enumsspb.NEXUS_OPERATION_STATE_BACKING_OFF,
324
enumsspb.NEXUS_OPERATION_STATE_STARTED,
325
},
326
enumsspb.NEXUS_OPERATION_STATE_SUCCEEDED,
328
>
// Not recording the last attempt information here since the state machine will be deleted immediately after this transition.
329
>
// If we ever use this code for a standalone state machine implementation we will want to record the last
330
>
// attempt information in case the completion is a result of a synchronous operation.
331
>
return op.output()
332
>
},
333
)
334
335
// EventCanceled is triggered when an invocation attempt succeeds.
336
type EventCanceled struct {
337
Time time.Time
338
Node *hsm.Node
339
}
340
341
var TransitionCanceled = hsm.NewTransition(
342
[]enumsspb.NexusOperationState{
343
enumsspb.NEXUS_OPERATION_STATE_SCHEDULED,
344
enumsspb.NEXUS_OPERATION_STATE_BACKING_OFF,
345
enumsspb.NEXUS_OPERATION_STATE_STARTED,
346
},
347
enumsspb.NEXUS_OPERATION_STATE_CANCELED,
349
>
// Not recording the last attempt information here since the state machine will be deleted immediately after this transition.
350
>
// If we ever use this code for a standalone state machine implementation we will want to record the last
351
>
// attempt information in case the completion is a result of a synchronous operation.
352
>
return op.output()
353
>
},
354
)
355
356
// EventStarted is triggered when an invocation attempt succeeds and the handler indicates that it started an
357
// asynchronous operation.
358
type EventStarted struct {
359
Time time.Time
360
Node *hsm.Node
361
Attributes *historypb.NexusOperationStartedEventAttributes
362
}
363
364
var TransitionStarted = hsm.NewTransition(
365
[]enumsspb.NexusOperationState{enumsspb.NEXUS_OPERATION_STATE_SCHEDULED, enumsspb.NEXUS_OPERATION_STATE_BACKING_OFF},
366
enumsspb.NEXUS_OPERATION_STATE_STARTED,
368
>
op.recordAttempt(event.Time)
369
>
if event.Attributes.OperationToken != "" {
371
>
} else if event.Attributes.OperationId != "" { //nolint:staticcheck // SA1019 this field might be set in older histories.
statemachine.go ×5
372
// TODO(bergundy): Remove this fallback after the 1.27 release.
373
op.OperationToken = event.Attributes.OperationId //nolint:staticcheck // SA1019 this field might be set in older histories.
374
}
375
377
>
378
>
// If cancelation is requested already, schedule sending the cancelation request.
379
>
child, err := op.CancelationNode(event.Node)
380
>
if err != nil {
381
return hsm.TransitionOutput{}, err
382
}
384
>
return hsm.TransitionOutput{}, hsm.MachineTransition(child, func(c Cancelation) (hsm.TransitionOutput, error) {
statemachine.go ×2
385
>
return TransitionCancelationScheduled.Apply(c, EventCancelationScheduled{
386
>
Time: event.Time,
387
>
Node: child,
388
>
})
389
>
})
390
}
391
393
>
if err != nil {
394
return output, err
395
}
396
397
// Schedule start-to-close timeout task if configured
399
>
400
>
return output, nil
401
},
402
)
403
404
// EventTimedOut is triggered when the schedule-to-close timeout is triggered for an operation.
405
type EventTimedOut struct {
406
Node *hsm.Node
407
}
408
409
var TransitionTimedOut = hsm.NewTransition(
410
[]enumsspb.NexusOperationState{
411
enumsspb.NEXUS_OPERATION_STATE_SCHEDULED,
412
enumsspb.NEXUS_OPERATION_STATE_BACKING_OFF,
413
enumsspb.NEXUS_OPERATION_STATE_STARTED,
414
},
415
enumsspb.NEXUS_OPERATION_STATE_TIMED_OUT,
417
>
// Keep attempt information as-is for debuggability.
418
>
// When used in a workflow, this machine node will be deleted from the tree after this transition.
419
>
return op.output()
420
>
},
421
)
422
423
// Cancel marks the Operation machine as canceled by spawning a child Cancelation machine. If the
424
// Operation already completed, then the Operation cannot be canceled anymore, and the Cancelation
425
// machine will stay in UNSPECIFIED state. If the Operation is in STARTED state, then transition the
426
// Cancelation machine to the SCHEDULED state. Otherwise, the Cancelation machine will wait the
427
// Operation machine transition to the STARTED state.
428
>
func (o Operation) Cancel(node *hsm.Node, t time.Time, requestedEventID int64) (hsm.TransitionOutput, error) {
statemachine.go ×3
429
>
child, err := node.AddChild(CancelationMachineKey, Cancelation{
430
>
NexusOperationCancellationInfo: &persistencespb.NexusOperationCancellationInfo{
431
>
RequestedEventId: requestedEventID,
432
>
},
433
>
})
434
>
if err != nil {
435
// This function should be called as part of command/event handling and it should not be called
436
// more than once.
437
return hsm.TransitionOutput{}, err
438
}
440
>
// Operation hasn't started yet or has already completed. Either way, cannot schedule
statemachine.go ×1
441
>
// cancelation.
442
>
return hsm.TransitionOutput{}, nil
443
>
}
444
>
return hsm.TransitionOutput{}, hsm.MachineTransition(child, func(c Cancelation) (hsm.TransitionOutput, error) {
statemachine.go ×1
445
>
return TransitionCancelationScheduled.Apply(c, EventCancelationScheduled{
446
>
Time: t,
447
>
Node: child,
448
>
})
449
>
})
450
}
451
452
// TODO: Remove this implementation once transition history is fully implemented.
454
>
switch o.State() {
455
case enumsspb.NEXUS_OPERATION_STATE_UNSPECIFIED:
456
return 0, 0, serviceerror.NewInvalidArgument("uninitialized operation state")
458
>
return 1, o.GetAttempt() * 2, nil
460
>
// We've made slightly more progress if we transitioned from backing off to scheduled.
461
>
return 1, o.GetAttempt()*2 + 1, nil
463
>
return 2, 0, nil
464
case enumsspb.NEXUS_OPERATION_STATE_TIMED_OUT,
465
enumsspb.NEXUS_OPERATION_STATE_FAILED,
466
enumsspb.NEXUS_OPERATION_STATE_CANCELED,
468
>
// Consider any terminal state as "max progress", we'll rely on last update namespace failover version to break
469
>
// the tie when comparing two states.
470
>
return terminalStage, 0, nil
471
default:
472
return 0, 0, serviceerror.NewInvalidArgument("unknown operation state")
473
}
474
}
475
476
type cancelationMachineDefinition struct{}
477
478
func (cancelationMachineDefinition) Deserialize(d []byte) (any, error) {
479
info := &persistencespb.NexusOperationCancellationInfo{}
480
return Cancelation{info}, proto.Unmarshal(d, info)
481
}
482
484
>
if state, ok := state.(Cancelation); ok {
485
>
return proto.Marshal(state.NexusOperationCancellationInfo)
486
>
}
487
return nil, fmt.Errorf("invalid cancelation provided: %v", state)
488
}
489
491
>
return CancelationMachineType
492
>
}
493
494
// CompareState compares the progress of two Cancelation state machines to determine whether to sync machine state while
495
// processing a replication task.
496
// TODO: Remove this implementation once transition history is fully implemented.
497
>
func (cancelationMachineDefinition) CompareState(state1, state2 any) (int, error) {
statemachine.go ×6
498
>
c1, ok := state1.(Cancelation)
499
>
if !ok {
500
return 0, fmt.Errorf("%w: expected state1 to be a Cancelation instance, got %v", hsm.ErrIncompatibleType, state1)
501
}
503
>
if !ok {
504
return 0, fmt.Errorf("%w: expected state2 to be a Cancelation instance, got %v", hsm.ErrIncompatibleType, state2)
505
}
506
508
>
if err != nil {
509
return 0, fmt.Errorf("failed to get progress for state1: %w", err)
510
}
512
>
if err != nil {
513
return 0, fmt.Errorf("failed to get progress for state2: %w", err)
514
}
517
>
}
519
>
return 0, serviceerror.NewInvalidArgumentf("cannot compare two distinct terminal states: %v, %v", c1.State(), c2.State())
statemachine.go ×1
520
>
}
522
}
523
524
// Cancelation state machine for canceling an operation.
525
type Cancelation struct {
526
*persistencespb.NexusOperationCancellationInfo
527
}
528
530
>
return c.NexusOperationCancellationInfo.State
531
>
}
532
533
>
func (c Cancelation) SetState(state enumspb.NexusOperationCancellationState) {
statemachine.go ×6
534
>
c.NexusOperationCancellationInfo.State = state
535
>
}
536
538
>
c.NexusOperationCancellationInfo.Attempt++
539
>
c.NexusOperationCancellationInfo.LastAttemptCompleteTime = timestamppb.New(ts)
540
>
c.NexusOperationCancellationInfo.LastAttemptFailure = nil
541
>
}
542
544
>
op, err := hsm.MachineData[Operation](node.Parent)
545
>
if err != nil {
546
return nil, err
547
}
549
>
case enumspb.NEXUS_OPERATION_CANCELLATION_STATE_SCHEDULED:
550
>
return []hsm.Task{CancelationTask{EndpointName: op.Endpoint, Attempt: c.Attempt}}, nil
552
>
return []hsm.Task{CancelationBackoffTask{deadline: c.NextAttemptScheduleTime.AsTime()}}, nil
554
>
return nil, nil
555
}
556
}
557
558
>
func (c Cancelation) output(node *hsm.Node) (hsm.TransitionOutput, error) {
statemachine.go ×6
559
>
tasks, err := c.RegenerateTasks(node)
560
>
if err != nil {
561
return hsm.TransitionOutput{}, err
562
}
564
}
565
566
// TODO: Remove this implementation once transition history is fully implemented.
568
>
switch c.State() {
569
case enumspb.NEXUS_OPERATION_CANCELLATION_STATE_UNSPECIFIED:
570
// UNSPECIFIED is a valid state since the cancelation may not initially get scheduled if the operation hasn't
571
// been started yet.
572
return 0, 0, nil
574
>
return 1, c.GetAttempt() * 2, nil
576
>
// We've made slightly more progress if we transitioned from backing off to scheduled.
577
>
return 1, c.GetAttempt()*2 + 1, nil
578
>
case enumspb.NEXUS_OPERATION_CANCELLATION_STATE_SUCCEEDED, enumspb.NEXUS_OPERATION_CANCELLATION_STATE_FAILED:
statemachine.go ×1
579
>
// Consider any terminal state as "max progress", we'll rely on last update namespace failover version to break
580
>
// the tie when comparing two states.
581
>
return terminalStage, 0, nil
582
default:
583
return 0, 0, serviceerror.NewInvalidArgument("unknown cancelation state")
584
}
585
}
586
587
// EventCancelationScheduled is triggered when cancelation is meant to be scheduled for the first time - immediately
588
// after it has been requested.
589
type EventCancelationScheduled struct {
590
Time time.Time
591
Node *hsm.Node
592
}
593
594
var TransitionCancelationScheduled = hsm.NewTransition(
595
[]enumspb.NexusOperationCancellationState{enumspb.NEXUS_OPERATION_CANCELLATION_STATE_UNSPECIFIED},
596
enumspb.NEXUS_OPERATION_CANCELLATION_STATE_SCHEDULED,
597
>
func(op Cancelation, event EventCancelationScheduled) (hsm.TransitionOutput, error) {
statemachine.go ×6
598
>
op.RequestedTime = timestamppb.New(event.Time)
599
>
return op.output(event.Node)
600
>
},
601
)
602
603
// EventCancelationRescheduled is triggered when cancelation is meant to be rescheduled after backing off from a
604
// previous attempt.
605
type EventCancelationRescheduled struct {
606
Node *hsm.Node
607
}
608
609
var TransitionCancelationRescheduled = hsm.NewTransition(
610
[]enumspb.NexusOperationCancellationState{enumspb.NEXUS_OPERATION_CANCELLATION_STATE_BACKING_OFF},
611
enumspb.NEXUS_OPERATION_CANCELLATION_STATE_SCHEDULED,
612
>
func(c Cancelation, event EventCancelationRescheduled) (hsm.TransitionOutput, error) {
statemachine.go ×1
613
>
c.NextAttemptScheduleTime = nil
614
>
return c.output(event.Node)
615
>
},
616
)
617
618
// EventCancelationAttemptFailed is triggered when a cancelation attempt is failed with a retryable error.
619
type EventCancelationAttemptFailed struct {
620
Time time.Time
621
Failure *failurepb.Failure
622
Node *hsm.Node
623
RetryPolicy backoff.RetryPolicy
624
}
625
626
var TransitionCancelationAttemptFailed = hsm.NewTransition(
627
[]enumspb.NexusOperationCancellationState{enumspb.NEXUS_OPERATION_CANCELLATION_STATE_SCHEDULED},
628
enumspb.NEXUS_OPERATION_CANCELLATION_STATE_BACKING_OFF,
629
>
func(c Cancelation, event EventCancelationAttemptFailed) (hsm.TransitionOutput, error) {
statemachine.go ×2
630
>
c.recordAttempt(event.Time)
631
>
// Use 0 for elapsed time as we don't limit the retry by time (for now).
632
>
nextDelay := event.RetryPolicy.ComputeNextDelay(0, int(c.Attempt), nil)
633
>
nextAttemptScheduleTime := event.Time.Add(nextDelay)
634
>
c.NextAttemptScheduleTime = timestamppb.New(nextAttemptScheduleTime)
635
>
c.LastAttemptFailure = event.Failure
636
>
return c.output(event.Node)
637
>
},
638
)
639
640
// EventCancelationFailed is triggered when a cancelation attempt is failed with a non retryable error.
641
type EventCancelationFailed struct {
642
Time time.Time
643
Failure *failurepb.Failure
644
Node *hsm.Node
645
}
646
647
var TransitionCancelationFailed = hsm.NewTransition(
648
[]enumspb.NexusOperationCancellationState{
649
// We can immediately transition to failed to since we don't know how to send a cancelation request for an
650
// unstarted operation.
651
enumspb.NEXUS_OPERATION_CANCELLATION_STATE_UNSPECIFIED,
652
enumspb.NEXUS_OPERATION_CANCELLATION_STATE_SCHEDULED,
653
},
654
enumspb.NEXUS_OPERATION_CANCELLATION_STATE_FAILED,
655
>
func(c Cancelation, event EventCancelationFailed) (hsm.TransitionOutput, error) {
statemachine.go ×1
656
>
c.recordAttempt(event.Time)
657
>
c.LastAttemptFailure = event.Failure
658
>
return c.output(event.Node)
659
>
},
660
)
661
662
// EventCancelationSucceeded is triggered when a cancelation attempt succeeds.
663
type EventCancelationSucceeded struct {
664
Time time.Time
665
Node *hsm.Node
666
}
667
668
var TransitionCancelationSucceeded = hsm.NewTransition(
669
[]enumspb.NexusOperationCancellationState{enumspb.NEXUS_OPERATION_CANCELLATION_STATE_SCHEDULED},
670
enumspb.NEXUS_OPERATION_CANCELLATION_STATE_SUCCEEDED,
671
>
func(c Cancelation, event EventCancelationSucceeded) (hsm.TransitionOutput, error) {
statemachine.go ×1
672
>
c.recordAttempt(event.Time)
673
>
return c.output(event.Node)
674
>
},
675
)
676
678
>
if err := r.RegisterMachine(operationMachineDefinition{}); err != nil {
679
return err
680
}
682
}