Vihren Atlas

go.temporal.io/server/components/nexusoperations/events.go

359 LOC · 156 covered · 203 uncovered · 57 ranges · 367 concepts · 27 introducers · 165 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.

Files sharing a concept 3

A concept that introduces code in this file also introduces code in these files.

Files with a broader (ancestor) concept 9

These files introduce code for a direct ancestor — a broader, larger-extent concept — of a concept that introduces code here.

Files with a narrower (descendant) concept 8

These files introduce code for a direct descendant — a narrower, larger-intent concept — of a concept that introduces code here.

1 package nexusoperations
2
3 import (
4 "fmt"
5 "reflect"
6 "strconv"
7
8 enumspb "go.temporal.io/api/enums/v1"
9 historypb "go.temporal.io/api/history/v1"
10 "go.temporal.io/server/service/history/hsm"
11 )
12
13 type ScheduledEventDefinition struct{}
14
15 func (d ScheduledEventDefinition) IsWorkflowTaskTrigger() bool {
16 return false
17 }
18
19 > func (d ScheduledEventDefinition) Type() enumspb.EventType { events.go ×13
20 > return enumspb.EVENT_TYPE_NEXUS_OPERATION_SCHEDULED
21 > }
22
23 > func (d ScheduledEventDefinition) Apply(root *hsm.Node, event *historypb.HistoryEvent) error { events.go ×2
24 > token, err := root.GenerateEventLoadToken(event)
25 > if err != nil {
26 return err
27 }
28 > _, err = AddChild(root, strconv.FormatInt(event.EventId, 10), event, token) events.go ×2
29 > return err
30 }
31
32 > func (d ScheduledEventDefinition) CherryPick(root *hsm.Node, event *historypb.HistoryEvent, _ map[enumspb.ResetReapplyExcludeType]struct{}) error { events.go ×6
33 > // We never cherry pick command events, and instead allow user logic to reschedule those commands.
34 > return hsm.ErrNotCherryPickable
35 > }
36
37 type CancelRequestedEventDefinition struct{}
38
39 func (d CancelRequestedEventDefinition) IsWorkflowTaskTrigger() bool {
40 return false
41 }
42
43 > func (d CancelRequestedEventDefinition) Type() enumspb.EventType { events.go ×13
44 > return enumspb.EVENT_TYPE_NEXUS_OPERATION_CANCEL_REQUESTED
45 > }
46
47 > func (d CancelRequestedEventDefinition) Apply(root *hsm.Node, event *historypb.HistoryEvent) error { commands.go ×4
48 > _, err := transitionOperation(root, event, func(node *hsm.Node, o Operation) (hsm.TransitionOutput, error) {
49 > return o.Cancel(node, event.EventTime.AsTime(), event.EventId) commands.go ×3
50 > })
51
52 > return err commands.go ×4
53 }
54
55 > func (d CancelRequestedEventDefinition) CherryPick(root *hsm.Node, event *historypb.HistoryEvent, _ map[enumspb.ResetReapplyExcludeType]struct{}) error { events.go ×6
56 > // We never cherry pick command events, and instead allow user logic to reschedule those commands.
57 > return hsm.ErrNotCherryPickable
58 > }
59
60 type CancelRequestCompletedEventDefinition struct{}
61
62 func (d CancelRequestCompletedEventDefinition) IsWorkflowTaskTrigger() bool {
63 return true
64 }
65
66 > func (d CancelRequestCompletedEventDefinition) Type() enumspb.EventType { events.go ×13
67 > return enumspb.EVENT_TYPE_NEXUS_OPERATION_CANCEL_REQUEST_COMPLETED
68 > }
69
70 func (d CancelRequestCompletedEventDefinition) Apply(root *hsm.Node, event *historypb.HistoryEvent) error {
71 _, err := transitionOperation(root, event, func(node *hsm.Node, o Operation) (hsm.TransitionOutput, error) {
72 child, err := o.CancelationNode(node)
73 if err != nil {
74 return hsm.TransitionOutput{}, err
75 }
76 if child != nil {
77 return hsm.TransitionOutput{}, hsm.MachineTransition(child, func(c Cancelation) (hsm.TransitionOutput, error) {
78 return TransitionCancelationSucceeded.Apply(c, EventCancelationSucceeded{
79 Time: event.EventTime.AsTime(),
80 Node: child,
81 })
82 })
83 }
84 return hsm.TransitionOutput{}, nil
85 })
86 return err
87 }
88
89 func (d CancelRequestCompletedEventDefinition) CherryPick(root *hsm.Node, event *historypb.HistoryEvent, _ map[enumspb.ResetReapplyExcludeType]struct{}) error {
90 // We never cherry pick command events, and instead allow user logic to reschedule those commands.
91 return hsm.ErrNotCherryPickable
92 }
93
94 type CancelRequestFailedEventDefinition struct{}
95
96 func (d CancelRequestFailedEventDefinition) IsWorkflowTaskTrigger() bool {
97 return true
98 }
99
100 > func (d CancelRequestFailedEventDefinition) Type() enumspb.EventType { events.go ×13
101 > return enumspb.EVENT_TYPE_NEXUS_OPERATION_CANCEL_REQUEST_FAILED
102 > }
103
104 func (d CancelRequestFailedEventDefinition) Apply(root *hsm.Node, event *historypb.HistoryEvent) error {
105 _, err := transitionOperation(root, event, func(node *hsm.Node, o Operation) (hsm.TransitionOutput, error) {
106 child, err := o.CancelationNode(node)
107 if err != nil {
108 return hsm.TransitionOutput{}, err
109 }
110 if child != nil {
111 return hsm.TransitionOutput{}, hsm.MachineTransition(child, func(c Cancelation) (hsm.TransitionOutput, error) {
112 return TransitionCancelationFailed.Apply(c, EventCancelationFailed{
113 Time: event.EventTime.AsTime(),
114 Failure: event.GetNexusOperationCancelRequestFailedEventAttributes().GetFailure(),
115 Node: child,
116 })
117 })
118 }
119 return hsm.TransitionOutput{}, nil
120 })
121 return err
122 }
123
124 func (d CancelRequestFailedEventDefinition) CherryPick(root *hsm.Node, event *historypb.HistoryEvent, _ map[enumspb.ResetReapplyExcludeType]struct{}) error {
125 // We never cherry pick command events, and instead allow user logic to reschedule those commands.
126 return hsm.ErrNotCherryPickable
127 }
128
129 type StartedEventDefinition struct{}
130
131 func (d StartedEventDefinition) IsWorkflowTaskTrigger() bool {
132 return true
133 }
134
135 > func (d StartedEventDefinition) Type() enumspb.EventType { events.go ×1
136 > return enumspb.EVENT_TYPE_NEXUS_OPERATION_STARTED
137 > }
138
139 > func (d StartedEventDefinition) Apply(root *hsm.Node, event *historypb.HistoryEvent) error { events.go ×2
140 > _, err := transitionOperation(root, event, func(node *hsm.Node, o Operation) (hsm.TransitionOutput, error) {
141 > return TransitionStarted.Apply(o, EventStarted{ events.go ×1
142 > Time: event.EventTime.AsTime(),
143 > Node: node,
144 > Attributes: event.GetNexusOperationStartedEventAttributes(),
145 > })
146 > })
147
148 > return err events.go ×2
149 }
150
151 > func (d StartedEventDefinition) CherryPick(root *hsm.Node, event *historypb.HistoryEvent, excludeTypes map[enumspb.ResetReapplyExcludeType]struct{}) error { events.go ×1
152 > if _, ok := excludeTypes[enumspb.RESET_REAPPLY_EXCLUDE_TYPE_NEXUS]; ok {
153 > return hsm.ErrNotCherryPickable events.go ×1
154 > }
155 > return d.Apply(root, event) events.go ×1
156 }
157
158 type CompletedEventDefinition struct{}
159
160 func (d CompletedEventDefinition) IsWorkflowTaskTrigger() bool {
161 return true
162 }
163
164 > func (d CompletedEventDefinition) Apply(root *hsm.Node, event *historypb.HistoryEvent) error { events.go ×2
165 > node, err := transitionOperation(root, event, func(node *hsm.Node, o Operation) (hsm.TransitionOutput, error) {
166 > return TransitionSucceeded.Apply(o, EventSucceeded{
167 > Time: event.EventTime.AsTime(),
168 > Node: node,
169 > })
170 > })
171 > if err != nil {
172 return err
173 }
174
175 > return node.Parent.DeleteChild(node.Key) events.go ×2
176 }
177
178 > func (d CompletedEventDefinition) Type() enumspb.EventType { events.go ×1
179 > return enumspb.EVENT_TYPE_NEXUS_OPERATION_COMPLETED
180 > }
181
182 > func (d CompletedEventDefinition) CherryPick(root *hsm.Node, event *historypb.HistoryEvent, excludeTypes map[enumspb.ResetReapplyExcludeType]struct{}) error { events.go ×6
183 > if _, ok := excludeTypes[enumspb.RESET_REAPPLY_EXCLUDE_TYPE_NEXUS]; ok {
184 > return hsm.ErrNotCherryPickable
185 > }
186 return d.Apply(root, event)
187 }
188
189 type FailedEventDefinition struct{}
190
191 func (d FailedEventDefinition) IsWorkflowTaskTrigger() bool {
192 return true
193 }
194
195 > func (d FailedEventDefinition) Type() enumspb.EventType { events.go ×1
196 > return enumspb.EVENT_TYPE_NEXUS_OPERATION_FAILED
197 > }
198
199 > func (d FailedEventDefinition) Apply(root *hsm.Node, event *historypb.HistoryEvent) error { events.go ×2
200 > node, err := transitionOperation(root, event, func(node *hsm.Node, o Operation) (hsm.TransitionOutput, error) {
201 > return TransitionFailed.Apply(o, EventFailed{
202 > Time: event.EventTime.AsTime(),
203 > Attributes: event.GetNexusOperationFailedEventAttributes(),
204 > Node: node,
205 > })
206 > })
207 > if err != nil {
208 return err
209 }
210
211 > return node.Parent.DeleteChild(node.Key) events.go ×2
212 }
213
214 > func (d FailedEventDefinition) CherryPick(root *hsm.Node, event *historypb.HistoryEvent, excludeTypes map[enumspb.ResetReapplyExcludeType]struct{}) error { events.go ×6
215 > if _, ok := excludeTypes[enumspb.RESET_REAPPLY_EXCLUDE_TYPE_NEXUS]; ok {
216 > return hsm.ErrNotCherryPickable
217 > }
218 return d.Apply(root, event)
219 }
220
221 type CanceledEventDefinition struct{}
222
223 func (d CanceledEventDefinition) IsWorkflowTaskTrigger() bool {
224 return true
225 }
226
227 > func (d CanceledEventDefinition) Type() enumspb.EventType { events.go ×1
228 > return enumspb.EVENT_TYPE_NEXUS_OPERATION_CANCELED
229 > }
230
231 > func (d CanceledEventDefinition) Apply(root *hsm.Node, event *historypb.HistoryEvent) error { events.go ×2
232 > node, err := transitionOperation(root, event, func(node *hsm.Node, o Operation) (hsm.TransitionOutput, error) {
233 > return TransitionCanceled.Apply(o, EventCanceled{
234 > Time: event.EventTime.AsTime(),
235 > Node: node,
236 > })
237 > })
238 > if err != nil {
239 return err
240 }
241
242 > return node.Parent.DeleteChild(node.Key) events.go ×2
243 }
244
245 > func (d CanceledEventDefinition) CherryPick(root *hsm.Node, event *historypb.HistoryEvent, excludeTypes map[enumspb.ResetReapplyExcludeType]struct{}) error { events.go ×6
246 > if _, ok := excludeTypes[enumspb.RESET_REAPPLY_EXCLUDE_TYPE_NEXUS]; ok {
247 > return hsm.ErrNotCherryPickable
248 > }
249 return d.Apply(root, event)
250 }
251
252 type TimedOutEventDefinition struct{}
253
254 func (d TimedOutEventDefinition) IsWorkflowTaskTrigger() bool {
255 return true
256 }
257
258 > func (d TimedOutEventDefinition) Type() enumspb.EventType { events.go ×1
259 > return enumspb.EVENT_TYPE_NEXUS_OPERATION_TIMED_OUT
260 > }
261
262 > func (d TimedOutEventDefinition) Apply(root *hsm.Node, event *historypb.HistoryEvent) error { events.go ×2
263 > node, err := transitionOperation(root, event, func(node *hsm.Node, o Operation) (hsm.TransitionOutput, error) {
264 > return TransitionTimedOut.Apply(o, EventTimedOut{
265 > Node: node,
266 > })
267 > })
268 > if err != nil {
269 return err
270 }
271
272 > return node.Parent.DeleteChild(node.Key) events.go ×2
273 }
274
275 > func (d TimedOutEventDefinition) CherryPick(root *hsm.Node, event *historypb.HistoryEvent, excludeTypes map[enumspb.ResetReapplyExcludeType]struct{}) error { events.go ×6
276 > if _, ok := excludeTypes[enumspb.RESET_REAPPLY_EXCLUDE_TYPE_NEXUS]; ok {
277 > return hsm.ErrNotCherryPickable
278 > }
279 return d.Apply(root, event)
280 }
281
282 > func RegisterEventDefinitions(reg *hsm.Registry) error { events.go ×13
283 > if err := reg.RegisterEventDefinition(ScheduledEventDefinition{}); err != nil {
284 return err
285 }
286 > if err := reg.RegisterEventDefinition(CancelRequestedEventDefinition{}); err != nil { events.go ×13
287 return err
288 }
289 > if err := reg.RegisterEventDefinition(CancelRequestCompletedEventDefinition{}); err != nil { events.go ×13
290 return err
291 }
292 > if err := reg.RegisterEventDefinition(CancelRequestFailedEventDefinition{}); err != nil { events.go ×13
293 return err
294 }
295 > if err := reg.RegisterEventDefinition(StartedEventDefinition{}); err != nil { events.go ×13
296 return err
297 }
298 > if err := reg.RegisterEventDefinition(CompletedEventDefinition{}); err != nil { events.go ×13
299 return err
300 }
301 > if err := reg.RegisterEventDefinition(FailedEventDefinition{}); err != nil { events.go ×13
302 return err
303 }
304 > if err := reg.RegisterEventDefinition(CanceledEventDefinition{}); err != nil { events.go ×13
305 return err
306 }
307 > return reg.RegisterEventDefinition(TimedOutEventDefinition{}) events.go ×13
308 }
309
310 func transitionOperation(
311 root *hsm.Node,
312 event *historypb.HistoryEvent,
313 fn func(node *hsm.Node, o Operation) (hsm.TransitionOutput, error),
314 > ) (*hsm.Node, error) { events.go ×4
315 > node, err := findOperationNode(root, event)
316 > if err != nil {
317 > return nil, err events.go ×1
318 > }
319 > if err := hsm.MachineTransition(node, func(o Operation) (hsm.TransitionOutput, error) { events.go ×3
320 > return fn(node, o)
321 > }); err != nil {
322 > return nil, err nexus.pb.go ×1
323 > }
324 > return node, nil events.go ×3
325 }
326
327 > func findOperationNode(root *hsm.Node, event *historypb.HistoryEvent) (*hsm.Node, error) { events.go ×4
328 > attrs := reflect.ValueOf(event.Attributes).Elem()
329 >
330 > // Attributes is always a struct with a single field (e.g: HistoryEvent_NexusOperationScheduledEventAttributes)
331 > if attrs.Kind() != reflect.Struct || attrs.NumField() != 1 {
332 panic("invalid event, expected Attributes field with a single field struct")
333 }
334
335 > f := attrs.Field(0).Interface() events.go ×4
336 >
337 > eventIDGetter, ok := f.(interface{ GetScheduledEventId() int64 })
338 > if !ok {
339 panic("Event does not have a ScheduledEventId field")
340 }
341 > coll := MachineCollection(root) events.go ×4
342 > nodeID := strconv.FormatInt(eventIDGetter.GetScheduledEventId(), 10)
343 > node, err := coll.Node(nodeID)
344 > if err != nil {
345 > return nil, err events.go ×1
346 > }
347 > requestIDGetter, ok := f.(interface{ GetRequestId() string }) events.go ×1
348 > if ok && requestIDGetter.GetRequestId() != "" {
349 > op, err := coll.Data(nodeID) events.go ×2
350 > if err != nil {
351 return nil, err
352 }
353 > if op.RequestId != requestIDGetter.GetRequestId() { events.go ×2
354 > return nil, fmt.Errorf("%w: event has different request ID (%q) than the machine (%q)", events.go ×1
355 > hsm.ErrNotCherryPickable, requestIDGetter.GetRequestId(), op.RequestId)
356 > }
357 }
358 > return node, nil events.go ×3
359 }