Vihren Atlas

completion.go ×18

Code frontier

unlabeled · c_65b2b3346156

3 tests · 3173 LOC · 139 files · introduces 0 tests · 57 LOC · 2 files

Introduces — evidence that enters the hierarchy at this concept

Code
20 ranges57 lines · 2 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
582 ranges3173 lines · 139 files · Browse complete extent
All tests (intent)
3 testsBrowse complete intent

Neighbourhood graph

The orange circle is the focus. Violet and green circles are every ancestor and descendant, broader and narrower, at any distance; blue squares and pink diamonds are the introduced files and exact introduced tests of every visible concept, not only the focus's. Arrows point from broader to narrower concepts and bridge only concepts omitted from this view. Undirected links show source or test introduction. Concept and file size follows LOC; exact test nodes use test-count units.

Introduced files, introduced tests, and structurally relevant concept specialization

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 native relationship evidence on this page.

Graph controls are ready.

Interactive rendering requires JavaScript and WebGL. Use the native relationship evidence on this page while the interactive map is unavailable.

Native relationship evidence

Every exact file and test below is linked only from the concept that introduces it.

Introduced tests

Every collected test enters the hierarchy at exactly one concept.

No tests are introduced at this concept. Its intent tests are introduced by other concepts.

Introduced code

Every collected source range enters the hierarchy at exactly one concept.

2 files ranked by introduced lines: 57 introduced LOC across 20 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/components/nexusoperations/completion.go 55 introduced LOC · 18 ranges

Open complete file

130 startTime *timestamppb.Timestamp,
131 links []*commonpb.Link,
132 > ) error { completion.go
133 > operation, err := hsm.MachineData[Operation](node)
134 > if err != nil {
135 return err
136 }
137
138 // The operation was already started, ignore.
139 > if !TransitionStarted.Possible(operation) { completion.go
140 return nil
141 }
142
143 > eventID, err := hsm.EventIDFromToken(operation.ScheduledEventToken) completion.go
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
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 })
163 > return (StartedEventDefinition{}).Apply(node.Parent, event) completion.go
164 }
165
189 result *commonpb.Payload,
190 opFailedError *nexus.OperationError,
191 > ) error { completion.go
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 }
203 > operation, err := hsm.MachineData[Operation](node) completion.go
204 > if err != nil {
205 return err
206 }
208 // fabricateStartedEventIfMissing will transition it to started below; the executor never emitted
209 // schedule-to-start in that case, so we emit it here.
210 > fabricatedStart := TransitionStarted.Possible(operation) completion.go
211 > if err := fabricateStartedEventIfMissing(node, requestID, operationToken, startTime, links); err != nil {
212 return err
213 }
214 > if requestID != "" && operation.RequestId != requestID { completion.go
215 isRetryableNotFoundErr = false
216 return serviceerror.NewNotFound("operation not found")
217 }
218
219 > if opFailedError != nil { completion.go
220 err = handleOperationError(node, operation, opFailedError)
221 > } else { completion.go
222 err = handleSuccessfulOperationResult(node, operation, result, nil)
223 }
224 // TODO(bergundy): Remove this once the operation auto-deletes itself from the tree on completion with state
225 // based replication.
226 > if errors.Is(err, hsm.ErrInvalidTransition) { completion.go
227 isRetryableNotFoundErr = false
228 return serviceerror.NewNotFound("operation not found")
229 }
230 > if err != nil { completion.go
231 return err
232 }
233 // fabricatedStart means the executor never emitted schedule-to-start, so we emit it here.
234 > emitScheduleToStart := fabricatedStart && operation.StartedTime != nil completion.go
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
239 // Try again without a run ID in case the original run was reset.
240 ref.WorkflowKey.RunID = ""
247 return h.Handle(ctx, env, ref, requestID, operationToken, startTime, links, result, opFailedError)
248 }
249 > if err != nil { completion.go
250 return err
251 }
252 > if emitMetrics != nil { completion.go
253 > emitMetrics()
254 > }
255 > return nil
256 }
257
265 emitScheduleToStart bool,
266 closeTime time.Time,
267 > ) func() { completion.go
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 {
275 case opFailedError == nil:
276 emitOperationSucceeded(metricsHandler, metricTagConfig, operation, namespaceName, workflowType, closeTime)
go.temporal.io/server/service/history/hsm/tree.go 2 introduced LOC · 2 ranges

Open complete file

507 return err
508 }
509 > if !execution.IsWorkflowExecutionRunning() { tree.go
510 return consts.ErrWorkflowCompleted
511 }
512 > return nil tree.go
513 }
514

Direct hierarchy

Parents

Children