go.temporal.io/server/components/callbacks/statemachine.go
273 LOC · 129 covered · 144 uncovered · 36 ranges · 245 concepts · 24 introducers · 90 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 callbacks
2
3
import (
4
"fmt"
5
"net/url"
6
"time"
7
8
enumspb "go.temporal.io/api/enums/v1"
9
failurepb "go.temporal.io/api/failure/v1"
10
"go.temporal.io/api/serviceerror"
11
enumsspb "go.temporal.io/server/api/enums/v1"
12
persistencespb "go.temporal.io/server/api/persistence/v1"
13
"go.temporal.io/server/common/backoff"
14
"go.temporal.io/server/common/persistence/serialization"
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
// StateMachineType is a unique type identifier for this state machine.
22
StateMachineType = "callbacks.Callback"
23
24
// A marker for the first return value from a progress() that indicates the machine is in a terminal state.
25
// TODO: Remove this once transition history is fully implemented.
26
terminalStage = 3
27
)
28
29
// MachineCollection creates a new typed [statemachines.Collection] for callbacks.
31
>
return hsm.NewCollection[Callback](tree, StateMachineType)
32
>
}
33
34
// Callback state machine.
35
//
36
// Deprecated: HSM Callback is no longer supported.
37
type Callback struct {
38
*persistencespb.CallbackInfo
39
}
40
41
// NewWorkflowClosedTrigger creates a WorkflowClosed trigger variant.
43
>
return &persistencespb.CallbackInfo_Trigger{
44
>
Variant: &persistencespb.CallbackInfo_Trigger_WorkflowClosed{},
45
>
}
46
>
}
47
48
// NewCallback creates a new callback in the STANDBY state from given params.
49
func NewCallback(
50
requestId string,
51
registrationTime *timestamppb.Timestamp,
52
trigger *persistencespb.CallbackInfo_Trigger,
53
cb *persistencespb.Callback,
55
>
return Callback{
56
>
&persistencespb.CallbackInfo{
57
>
Trigger: trigger,
58
>
Callback: cb,
59
>
State: enumsspb.CALLBACK_STATE_STANDBY,
60
>
RegistrationTime: registrationTime,
61
>
RequestId: requestId,
62
>
},
63
>
}
64
>
}
65
67
>
return c.CallbackInfo.State
68
>
}
69
71
>
c.CallbackInfo.State = state
72
>
}
73
75
>
c.CallbackInfo.Attempt++
76
>
c.CallbackInfo.LastAttemptCompleteTime = timestamppb.New(ts)
77
>
}
78
80
>
switch c.CallbackInfo.State {
82
>
return []hsm.Task{BackoffTask{deadline: c.NextAttemptScheduleTime.AsTime()}}, nil
84
>
switch v := c.Callback.GetVariant().(type) {
85
>
case *persistencespb.Callback_Nexus_:
86
>
u, err := url.Parse(c.Callback.GetNexus().Url)
87
>
if err != nil {
88
return nil, fmt.Errorf("failed to parse URL: %v: %w", &c, err)
89
}
90
>
return []hsm.Task{InvocationTask{destination: u.Scheme + "://" + u.Host}}, nil
statemachine.go ×2
91
case *persistencespb.Callback_Hsm:
92
// Destination is empty on the internal queue.
93
return []hsm.Task{InvocationTask{"TODO(bergundy): make this empty"}}, nil
94
95
default:
96
return nil, fmt.Errorf("unsupported callback variant %v", v)
97
}
98
}
100
}
101
103
>
// Task logic is the same when regenerating tasks for a given state and when transitioning to that state.
104
>
// Node is ignored here.
105
>
tasks, err := c.RegenerateTasks(nil)
106
>
return hsm.TransitionOutput{Tasks: tasks}, err
107
>
}
108
109
// TODO: Remove this implementation once transition history is fully implemented.
111
>
switch c.State() {
112
case enumsspb.CALLBACK_STATE_UNSPECIFIED:
113
return 0, 0, serviceerror.NewInvalidArgument("uninitialized callback state")
115
>
return 1, 0, nil
117
>
return 2, c.GetAttempt() * 2, nil
119
>
// We've made slightly more progress if we transitioned from backing off to scheduled.
120
>
return 2, c.GetAttempt()*2 + 1, nil
122
>
// Consider any terminal state as "max progress", we'll rely on last update namespace failover version to break
123
>
// the tie when comparing two states.
124
>
return terminalStage, 0, nil
125
default:
126
return 0, 0, serviceerror.NewInvalidArgument("unknown callback state")
127
}
128
}
129
130
type stateMachineDefinition struct{}
131
133
>
return StateMachineType
134
>
}
135
136
>
func (stateMachineDefinition) Deserialize(d []byte) (any, error) {
statemachine_environment.go ×7
137
>
info := &persistencespb.CallbackInfo{}
138
>
if err := proto.Unmarshal(d, info); err != nil {
139
return nil, serialization.NewDeserializationError(enumspb.ENCODING_TYPE_PROTO3, err)
140
}
142
}
143
145
>
if state, ok := state.(Callback); ok {
146
>
return proto.Marshal(state.CallbackInfo)
147
>
}
148
return nil, fmt.Errorf("invalid callback provided: %v", state)
149
}
150
151
// CompareState compares the progress of two Callback state machines to determine whether to sync machine state while
152
// processing a replication task.
153
// TODO: Remove this implementation once transition history is fully implemented.
154
>
func (stateMachineDefinition) CompareState(state1, state2 any) (int, error) {
statemachine.go ×6
155
>
cb1, ok := state1.(Callback)
156
>
if !ok {
157
return 0, fmt.Errorf("%w: expected state1 to be a Callback instance, got %v", hsm.ErrIncompatibleType, state1)
158
}
160
>
if !ok {
161
return 0, fmt.Errorf("%w: expected state2 to be a Callback instance, got %v", hsm.ErrIncompatibleType, state2)
162
}
163
165
>
if err != nil {
166
return 0, fmt.Errorf("failed to get progress for state1: %w", err)
167
}
169
>
if err != nil {
170
return 0, fmt.Errorf("failed to get progress for state2: %w", err)
171
}
174
>
}
176
>
return 0, serviceerror.NewInvalidArgumentf("cannot compare two distinct terminal states: %v, %v", cb1.State(), cb2.State())
common.pb.go ×2
177
>
}
179
}
180
182
>
return r.RegisterMachine(stateMachineDefinition{})
183
>
}
184
185
// EventScheduled is triggered when the callback is meant to be scheduled for the first time - when its Trigger
186
// condition is met.
187
type EventScheduled struct{}
188
189
var TransitionScheduled = hsm.NewTransition(
190
[]enumsspb.CallbackState{enumsspb.CALLBACK_STATE_STANDBY},
191
enumsspb.CALLBACK_STATE_SCHEDULED,
193
>
return cb.output()
194
>
},
195
)
196
197
// EventRescheduled is triggered when the callback is meant to be rescheduled after backing off from a previous attempt.
198
type EventRescheduled struct{}
199
200
var TransitionRescheduled = hsm.NewTransition(
201
[]enumsspb.CallbackState{enumsspb.CALLBACK_STATE_BACKING_OFF},
202
enumsspb.CALLBACK_STATE_SCHEDULED,
204
>
cb.CallbackInfo.NextAttemptScheduleTime = nil
205
>
return cb.output()
206
>
},
207
)
208
209
// EventAttemptFailed is triggered when an attempt is failed with a retryable error.
210
type EventAttemptFailed struct {
211
Time time.Time
212
Err error
213
RetryPolicy backoff.RetryPolicy
214
}
215
216
var TransitionAttemptFailed = hsm.NewTransition(
217
[]enumsspb.CallbackState{enumsspb.CALLBACK_STATE_SCHEDULED},
218
enumsspb.CALLBACK_STATE_BACKING_OFF,
219
>
func(cb Callback, event EventAttemptFailed) (hsm.TransitionOutput, error) {
statemachine.go ×2
220
>
cb.recordAttempt(event.Time)
221
>
// Use 0 for elapsed time as we don't limit the retry by time (for now).
222
>
nextDelay := event.RetryPolicy.ComputeNextDelay(0, int(cb.Attempt), event.Err)
223
>
nextAttemptScheduleTime := event.Time.Add(nextDelay)
224
>
cb.CallbackInfo.NextAttemptScheduleTime = timestamppb.New(nextAttemptScheduleTime)
225
>
cb.CallbackInfo.LastAttemptFailure = &failurepb.Failure{
226
>
Message: event.Err.Error(),
227
>
FailureInfo: &failurepb.Failure_ApplicationFailureInfo{
228
>
ApplicationFailureInfo: &failurepb.ApplicationFailureInfo{
229
>
NonRetryable: false,
230
>
},
231
>
},
232
>
}
233
>
return cb.output()
234
>
},
235
)
236
237
// EventFailed is triggered when an attempt is failed with a non retryable error.
238
type EventFailed struct {
239
Time time.Time
240
Err error
241
}
242
243
var TransitionFailed = hsm.NewTransition(
244
[]enumsspb.CallbackState{enumsspb.CALLBACK_STATE_SCHEDULED},
245
enumsspb.CALLBACK_STATE_FAILED,
247
>
cb.recordAttempt(event.Time)
248
>
cb.CallbackInfo.LastAttemptFailure = &failurepb.Failure{
249
>
Message: event.Err.Error(),
250
>
FailureInfo: &failurepb.Failure_ApplicationFailureInfo{
251
>
ApplicationFailureInfo: &failurepb.ApplicationFailureInfo{
252
>
NonRetryable: true,
253
>
},
254
>
},
255
>
}
256
>
return cb.output()
257
>
},
258
)
259
260
// EventSucceeded is triggered when an attempt succeeds.
261
type EventSucceeded struct {
262
Time time.Time
263
}
264
265
var TransitionSucceeded = hsm.NewTransition(
266
[]enumsspb.CallbackState{enumsspb.CALLBACK_STATE_SCHEDULED},
267
enumsspb.CALLBACK_STATE_SUCCEEDED,
269
>
cb.recordAttempt(event.Time)
270
>
cb.CallbackInfo.LastAttemptFailure = nil
271
>
return cb.output()
272
>
},
273
)