Vihren Atlas

executors.go ×16

Code frontier

unlabeled · c_47007857a411

19 tests · 3158 LOC · 142 files · introduces 0 tests · 62 LOC · 1 file

Introduces — evidence that enters the hierarchy at this concept

Code
16 ranges62 lines · 1 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
558 ranges3158 lines · 142 files · Browse complete extent
All tests (intent)
19 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.

1 file ranked by introduced lines: 62 introduced LOC across 16 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/components/nexusoperations/executors.go 62 introduced LOC · 16 ranges

Open complete file

164 }
165
166 > func (e taskExecutor) executeInvocationTask(ctx context.Context, env hsm.Environment, ref hsm.Ref, task InvocationTask) error { executors.go
167 > ns, err := e.NamespaceRegistry.GetNamespaceByID(namespace.ID(ref.WorkflowKey.NamespaceID))
168 > if err != nil {
169 return fmt.Errorf("failed to get namespace by ID: %w", err)
170 }
171 > args, err := e.loadOperationArgs(ctx, ns, env, ref) executors.go
172 > if err != nil {
173 return fmt.Errorf("failed to load operation args: %w", err)
174 }
175 > var endpoint *persistencespb.NexusEndpointEntry executors.go
176 >
177 > // Skip endpoint lookup for system-internal operations.
178 > if args.endpointName != commonnexus.SystemEndpoint {
179 // This happens when we accept the ScheduleNexusOperation command when the endpoint is not found in the registry as
180 // indicated by the EndpointNotFoundAlwaysNonRetryable dynamic config.
371 env hsm.Environment,
372 ref hsm.Ref,
373 > ) (args startArgs, err error) { executors.go
374 > var eventToken []byte
375 > err = env.Access(ctx, ref, hsm.AccessRead, func(node *hsm.Node) error {
376 > operation, err := hsm.MachineData[Operation](node)
377 > if err != nil {
378 return err
379 }
380
381 > args.endpointName = operation.Endpoint executors.go
382 > args.endpointID = operation.EndpointId
383 > args.service = operation.Service
384 > args.operation = operation.Operation
385 > args.requestID = operation.RequestId
386 > args.scheduledTime = operation.ScheduledTime.AsTime()
387 > args.scheduleToCloseTimeout = operation.ScheduleToCloseTimeout.AsDuration()
388 > args.scheduleToStartTimeout = operation.ScheduleToStartTimeout.AsDuration()
389 > args.startToCloseTimeout = operation.StartToCloseTimeout.AsDuration()
390 > eventToken = operation.ScheduledEventToken
391 > event, err := node.LoadHistoryEvent(ctx, eventToken)
392 > if err != nil {
393 return err
394 }
395 > attrs := event.GetNexusOperationScheduledEventAttributes() executors.go
396 > args.payload = attrs.GetInput()
397 > args.header = maps.Clone(attrs.GetNexusHeader())
398 > args.nexusLink = commonnexus.ConvertLinkWorkflowEventToNexusLink(&commonpb.Link_WorkflowEvent{
399 > Namespace: ns.Name().String(),
400 > WorkflowId: ref.WorkflowKey.WorkflowID,
401 > RunId: ref.WorkflowKey.RunID,
402 > Reference: &commonpb.Link_WorkflowEvent_EventRef{
403 > EventRef: &commonpb.Link_WorkflowEvent_EventReference{
404 > EventId: event.GetEventId(),
405 > EventType: event.GetEventType(),
406 > },
407 > },
408 > })
409 > args.namespaceFailoverVersion = event.Version
410 > return nil
411 })
412 > return executors.go
413 }
414
415 > func (e taskExecutor) saveResult(ctx context.Context, env hsm.Environment, ref hsm.Ref, result *nexusrpc.ClientStartOperationResponse[*commonpb.Payload], callErr error) error { executors.go
416 > // emitMetrics is derived from the operation's resulting state inside the Access closure and
417 > // invoked only after the write transaction commits successfully, so a failed commit (which
418 > // retries the task) does not double-count the metric. See operationMetricsHandler's doc comment.
419 > var emitMetrics func()
420 > err := env.Access(ctx, ref, hsm.AccessWrite, func(node *hsm.Node) error {
421 > operation, err := hsm.MachineData[Operation](node)
422 > if err != nil {
423 return err
424 }
425 > switch { executors.go
426 case callErr != nil:
427 err = e.handleStartOperationError(env, node, operation, callErr)
433 err = handleSuccessfulOperationResult(node, operation, result.Successful, links)
434 }
435 > if err != nil { executors.go
436 return err
437 }
438 // Derive the metric from the resulting state (set by the transition above) rather than from
439 // each branch, mirroring how the completion handler emits from the post-transition state.
440 > finalOp, err := hsm.MachineData[Operation](node) executors.go
441 > if err != nil {
442 return err
443 }
444 > emitMetrics = e.deferredOperationMetric(finalOp, callErr, node.NamespaceName(), node.WorkflowTypeName(), env.Now()) executors.go
445 > return nil
446 })
447 > if err != nil { executors.go
448 return err
449 }
450 > if emitMetrics != nil { executors.go
451 emitMetrics()
452 }
453 > return nil executors.go
454 }
455
490 // commits so the metric is not double-counted if the commit fails and the task is retried. callErr
491 // carries the timeout type for the below-min-request-timeout case.
492 > func (e taskExecutor) deferredOperationMetric(op Operation, callErr error, namespaceName, workflowType string, closeTime time.Time) func() { executors.go
493 > switch op.State() {
494 case enumsspb.NEXUS_OPERATION_STATE_SUCCEEDED:
495 return func() {

Direct hierarchy

Parents

Children