go.temporal.io/server/components/nexusoperations/completion.go
283 LOC · 135 covered · 148 uncovered · 36 ranges · 81 concepts · 10 introducers · 20 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
"context"
5
"errors"
6
"fmt"
7
"time"
8
9
"github.com/nexus-rpc/sdk-go/nexus"
10
commonpb "go.temporal.io/api/common/v1"
11
enumspb "go.temporal.io/api/enums/v1"
12
failurepb "go.temporal.io/api/failure/v1"
13
historypb "go.temporal.io/api/history/v1"
14
"go.temporal.io/api/serviceerror"
15
"go.temporal.io/server/common/metrics"
16
commonnexus "go.temporal.io/server/common/nexus"
17
"go.temporal.io/server/service/history/hsm"
18
"google.golang.org/protobuf/types/known/timestamppb"
19
)
20
21
func handleSuccessfulOperationResult(
22
node *hsm.Node,
23
operation Operation,
24
result *commonpb.Payload,
25
links []*commonpb.Link,
27
>
eventID, err := hsm.EventIDFromToken(operation.ScheduledEventToken)
28
>
if err != nil {
29
return err
30
}
31
>
event := node.AddHistoryEvent(enumspb.EVENT_TYPE_NEXUS_OPERATION_COMPLETED, func(e *historypb.HistoryEvent) {
completion.go ×2
32
>
// We must assign to this property, linter doesn't like this.
33
>
// nolint:revive
34
>
e.Attributes = &historypb.HistoryEvent_NexusOperationCompletedEventAttributes{
35
>
NexusOperationCompletedEventAttributes: &historypb.NexusOperationCompletedEventAttributes{
36
>
ScheduledEventId: eventID,
37
>
Result: result,
38
>
RequestId: operation.RequestId,
39
>
},
40
>
}
41
>
e.Links = links
42
>
})
43
>
return CompletedEventDefinition{}.Apply(node.Parent, event)
44
}
45
46
func handleOperationError(
47
node *hsm.Node,
48
operation Operation,
49
opErr *nexus.OperationError,
51
>
eventID, err := hsm.EventIDFromToken(operation.ScheduledEventToken)
52
>
if err != nil {
53
return err
54
}
56
>
// Special marker for Temporal->Temporal calls to indicate that the original failure should be unwrapped.
57
>
// Temporal uses a wrapper operation error with no additional information to transmit the OperationError over the network.
58
>
// The meaningful information is in the operation error's cause.
59
>
unwrapError := opErr.OriginalFailure.Metadata["unwrap-error"] == "true"
60
>
61
>
if unwrapError && opErr.OriginalFailure.Cause != nil {
62
var err error
63
originalCause, err = commonnexus.NexusFailureToTemporalFailure(*opErr.OriginalFailure.Cause)
64
if err != nil {
65
return serviceerror.NewInvalidArgumentf("Malformed failure: %v", err)
66
}
68
>
// Transform the OperationError to either ApplicationFailure or CanceledFailure based on the operation error state.
69
>
originalCause, err = commonnexus.NexusFailureToTemporalFailure(*opErr.OriginalFailure)
70
>
if err != nil {
71
return serviceerror.NewInvalidArgumentf("Malformed failure: %v", err)
72
}
73
}
74
77
>
event := node.AddHistoryEvent(enumspb.EVENT_TYPE_NEXUS_OPERATION_FAILED, func(e *historypb.HistoryEvent) {
78
>
// We must assign to this property, linter doesn't like this.
79
>
// nolint:revive
80
>
e.Attributes = &historypb.HistoryEvent_NexusOperationFailedEventAttributes{
81
>
NexusOperationFailedEventAttributes: &historypb.NexusOperationFailedEventAttributes{
82
>
Failure: createNexusOperationFailure(operation, eventID, originalCause),
83
>
ScheduledEventId: eventID,
84
>
RequestId: operation.RequestId,
85
>
},
86
>
}
87
>
})
88
91
>
if originalCause.GetCanceledFailureInfo() == nil {
93
>
originalCause = &failurepb.Failure{
94
>
Message: originalCause.GetMessage(),
95
>
StackTrace: originalCause.GetStackTrace(),
96
>
FailureInfo: &failurepb.Failure_CanceledFailureInfo{
97
>
CanceledFailureInfo: &failurepb.CanceledFailureInfo{},
98
>
},
99
>
Cause: originalCause.GetCause(),
100
>
}
101
>
}
102
>
event := node.AddHistoryEvent(enumspb.EVENT_TYPE_NEXUS_OPERATION_CANCELED, func(e *historypb.HistoryEvent) {
completion.go ×3
103
>
// We must assign to this property, linter doesn't like this.
104
>
// nolint:revive
105
>
e.Attributes = &historypb.HistoryEvent_NexusOperationCanceledEventAttributes{
106
>
NexusOperationCanceledEventAttributes: &historypb.NexusOperationCanceledEventAttributes{
107
>
Failure: createNexusOperationFailure(operation, eventID, originalCause),
108
>
ScheduledEventId: eventID,
109
>
RequestId: operation.RequestId,
110
>
},
111
>
}
112
>
})
113
115
default:
116
// Both the Nexus Client and CompletionHandler reject invalid states, but just in case, we return this as a
117
// transition error.
118
return fmt.Errorf("unexpected operation state: %v", opErr.State)
119
}
120
}
121
122
// fabricateStartedEventIfMissing adds a NEXUS_OPERATION_STARTED history event and transitions the
123
// operation to NEXUS_OPERATION_STATE_STARTED if it has not started yet. It is necessary if the
124
// completion is received before the start response. Callers that need to know whether a start was
125
// fabricated should check the operation's state before calling (see TransitionStarted.Possible).
126
func fabricateStartedEventIfMissing(
127
node *hsm.Node,
128
requestID string,
129
operationToken string,
130
startTime *timestamppb.Timestamp,
131
links []*commonpb.Link,
133
>
operation, err := hsm.MachineData[Operation](node)
134
>
if err != nil {
135
return err
136
}
137
138
// The operation was already started, ignore.
140
return nil
141
}
142
144
>
if err != nil {
145
return err
146
}
147
148
>
event := node.AddHistoryEvent(enumspb.EVENT_TYPE_NEXUS_OPERATION_STARTED, func(e *historypb.HistoryEvent) {
completion.go ×18
149
>
e.Attributes = &historypb.HistoryEvent_NexusOperationStartedEventAttributes{
150
>
NexusOperationStartedEventAttributes: &historypb.NexusOperationStartedEventAttributes{
151
>
ScheduledEventId: eventID,
152
>
OperationToken: operationToken,
153
>
// TODO(bergundy): Remove this fallback after the 1.27 release.
154
>
OperationId: operationToken,
155
>
RequestId: requestID,
156
>
},
157
>
}
158
>
e.Links = links
159
>
if startTime != nil {
160
e.EventTime = startTime
161
}
162
})
164
}
165
166
// CompletionHandler resolves async Nexus operation completions delivered to the history service
167
// and emits the caller-side terminal metrics. It is provided via fx and injected into the history
168
// handler so the metrics handler and tag config are sourced from the dependency graph rather than
169
// threaded through the call site.
170
type CompletionHandler struct {
171
metricsHandler metrics.Handler
172
config *Config
173
}
174
175
// NewCompletionHandler returns a CompletionHandler. Wired via fx; see Module.
176
>
func NewCompletionHandler(metricsHandler metrics.Handler, config *Config) *CompletionHandler {
completion.go ×1
177
>
return &CompletionHandler{metricsHandler: metricsHandler, config: config}
178
>
}
179
180
// Handle resolves an async Nexus operation completion.
181
func (h *CompletionHandler) Handle(
182
ctx context.Context,
183
env hsm.Environment,
184
ref hsm.Ref,
185
requestID string,
186
operationToken string,
187
startTime *timestamppb.Timestamp,
188
links []*commonpb.Link,
189
result *commonpb.Payload,
190
opFailedError *nexus.OperationError,
192
>
// The initial version of the completion token did not include a request ID.
193
>
// Only retry Access without a run ID if the request ID is not empty.
194
>
isRetryableNotFoundErr := requestID != ""
195
>
// emitMetrics is populated inside the Access closure and invoked only after the write
196
>
// transaction commits successfully, so a failed commit (which retries the completion) does not
197
>
// double-count the metric. See operationMetricsHandler's doc comment.
198
>
var emitMetrics func()
199
>
err := env.Access(ctx, ref, hsm.AccessWrite, func(node *hsm.Node) error {
200
>
if err := node.CheckRunning(); err != nil {
201
return err
202
}
204
>
if err != nil {
205
return err
206
}
207
// If the operation has not started yet, this completion arrived before the start response and
208
// fabricateStartedEventIfMissing will transition it to started below; the executor never emitted
209
// schedule-to-start in that case, so we emit it here.
211
>
if err := fabricateStartedEventIfMissing(node, requestID, operationToken, startTime, links); err != nil {
212
return err
213
}
215
isRetryableNotFoundErr = false
216
return serviceerror.NewNotFound("operation not found")
217
}
218
223
>
}
224
// TODO(bergundy): Remove this once the operation auto-deletes itself from the tree on completion with state
225
// based replication.
227
isRetryableNotFoundErr = false
228
return serviceerror.NewNotFound("operation not found")
229
}
231
return err
232
}
233
// fabricatedStart means the executor never emitted schedule-to-start, so we emit it here.
235
>
emitMetrics = h.deferredCompletionMetric(operation, node.NamespaceName(), node.WorkflowTypeName(), opFailedError, emitScheduleToStart, env.Now())
236
>
return nil
237
})
238
>
if errors.As(err, new(*serviceerror.NotFound)) && isRetryableNotFoundErr && ref.WorkflowKey.RunID != "" {
completion.go ×18
239
// Try again without a run ID in case the original run was reset.
240
ref.WorkflowKey.RunID = ""
241
// VersionedTransition is for a specific run. After reset, the TransitionCount will
242
// start from 1 again. Reset the TransitionCount to 0 here to fallback to old ref
243
// validation logic.
244
ref.StateMachineRef.MutableStateVersionedTransition = nil
245
ref.StateMachineRef.MachineInitialVersionedTransition.TransitionCount = 0
246
ref.StateMachineRef.MachineLastUpdateVersionedTransition.TransitionCount = 0
247
return h.Handle(ctx, env, ref, requestID, operationToken, startTime, links, result, opFailedError)
248
}
250
return err
251
}
253
>
emitMetrics()
254
>
}
255
>
return nil
256
}
257
258
// deferredCompletionMetric builds the post-commit caller-side emit for a resolved async completion:
259
// the terminal outcome (the canceled vs failed split mirrors handleOperationError so the metric
260
// matches the recorded transition) plus, when the start was fabricated here, schedule-to-start.
261
func (h *CompletionHandler) deferredCompletionMetric(
262
operation Operation,
263
namespaceName, workflowType string,
264
opFailedError *nexus.OperationError,
265
emitScheduleToStart bool,
266
closeTime time.Time,
268
>
metricsHandler := h.metricsHandler
269
>
metricTagConfig := h.config.ResolvedMetricTagConfig()
270
>
return func() {
271
>
if emitScheduleToStart {
272
>
emitScheduleToStartLatency(metricsHandler, metricTagConfig, operation, namespaceName, workflowType, operation.StartedTime.AsTime())
273
>
}
274
>
switch {
276
>
emitOperationSucceeded(metricsHandler, metricTagConfig, operation, namespaceName, workflowType, closeTime)
278
>
emitOperationCanceled(metricsHandler, metricTagConfig, operation, namespaceName, workflowType, closeTime)
280
>
emitOperationFailed(metricsHandler, metricTagConfig, operation, namespaceName, workflowType, closeTime)
281
}
282
}
283
}