Vihren Atlas

invoker_tasks.go ×7

Code frontier

unlabeled · c_92dfd8fd0ccc

6 tests · 5828 LOC · 180 files · introduces 0 tests · 89 LOC · 4 files

Introduces — evidence that enters the hierarchy at this concept

Code
15 ranges89 lines · 4 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
1341 ranges5828 lines · 180 files · Browse complete extent
All tests (intent)
6 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.

4 files ranked by introduced lines: 89 introduced LOC across 15 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/chasm/lib/scheduler/invoker_tasks.go 41 introduced LOC · 7 ranges

Open complete file

628
629 // Get rate limiter permission once per buffered start, on the first attempt only.
630 > if start.Attempt == 1 { invoker_tasks.go
631 > delay, err := h.getRateLimiterPermission()
632 > if err != nil {
633 return err
634 }
635 > if delay > 0 { invoker_tasks.go
636 return newRateLimitedError(delay)
637 }
638 }
639
640 > reusePolicy := enumspb.WORKFLOW_ID_REUSE_POLICY_REJECT_DUPLICATE invoker_tasks.go
641 > if start.Manual {
642 reusePolicy = enumspb.WORKFLOW_ID_REUSE_POLICY_ALLOW_DUPLICATE
643 }
644
645 > var lcr []*commonpb.Payload invoker_tasks.go
646 > if lastCompletionState.Success != nil {
647 lcr = append(lcr, lastCompletionState.Success)
648 }
651 // continue-as-new, rather than the started workflow's callback state which is re-stamped on each
652 // new run.
653 > callback, err := chasm.GenerateNexusCallback(schedulerRef, start.RequestId, h.config.EncodeInternalTokenWithEnvelope(scheduler.Namespace)) invoker_tasks.go
654 > if err != nil {
655 return err
656 }
657 > request := &workflowservice.StartWorkflowExecutionRequest{ invoker_tasks.go
658 > CompletionCallbacks: []*commonpb.Callback{callback},
659 > Header: requestSpec.Header,
660 > Identity: scheduler.identity(),
661 > Input: requestSpec.Input,
662 > Memo: requestSpec.Memo,
663 > Namespace: scheduler.Namespace,
664 > RequestId: start.RequestId,
665 > RetryPolicy: requestSpec.RetryPolicy,
666 > SearchAttributes: scheduler.startWorkflowSearchAttributes(start.NominalTime.AsTime()),
667 > TaskQueue: requestSpec.TaskQueue,
668 > UserMetadata: requestSpec.UserMetadata,
669 > WorkflowExecutionTimeout: requestSpec.WorkflowExecutionTimeout,
670 > WorkflowId: start.WorkflowId,
671 > WorkflowIdReusePolicy: reusePolicy,
672 > WorkflowRunTimeout: requestSpec.WorkflowRunTimeout,
673 > WorkflowTaskTimeout: requestSpec.WorkflowTaskTimeout,
674 > WorkflowType: requestSpec.WorkflowType,
675 > Priority: requestSpec.Priority,
676 > ContinuedFailure: lastCompletionState.Failure,
677 > LastCompletionResult: &commonpb.Payloads{
678 > Payloads: lcr,
679 > },
680 > }
681 >
682 > result, err := h.frontendClient.StartWorkflowExecution(ctx, request)
683 > if err != nil {
684 return err
685 }
748 // before proceeding. If an error is returned, execution should not proceed, and
749 // reservation should be retried.
750 > func (h *InvokerExecuteTaskHandler) getRateLimiterPermission() (delay time.Duration, err error) { invoker_tasks.go
751 > // For now, we're only going to rate limit via APS.
752 > return
753 > }
754
755 func isAlreadyStartedError(err error) bool {
go.temporal.io/server/chasm/nexus_completion.go 21 introduced LOC · 4 ranges

Open complete file

31 // config) to keep the envelope format off the wire until the whole fleet can read it. Either format is
32 // always decodable by UnpackNexusCallbackToken.
33 > func GenerateNexusCallback(serializedRef []byte, requestID string, encodeToken bool) (*commonpb.Callback, error) { nexus_completion.go
34 > var token string
35 > if encodeToken {
36 > var err error
37 > token, err = packNexusCallbackToken(serializedRef, requestID)
38 > if err != nil {
39 return nil, err
40 }
43 token = base64.RawURLEncoding.EncodeToString(serializedRef)
44 }
45 > return &commonpb.Callback{ nexus_completion.go
46 > Variant: &commonpb.Callback_Nexus_{
47 > Nexus: &commonpb.Callback_Nexus{
48 > Url: NexusCompletionHandlerURL,
49 > Header: map[string]string{nexusCallbackTokenHeader: token},
50 > },
51 > },
52 > }, nil
53 }
54
55 // packNexusCallbackToken encodes a CHASM component ref and request ID into a callback token.
56 > func packNexusCallbackToken(componentRef []byte, requestID string) (string, error) { nexus_completion.go
57 > b, err := proto.Marshal(&tokenspb.NexusOperationCompletion{
58 > ComponentRef: componentRef,
59 > RequestId: requestID,
60 > })
61 > if err != nil {
62 return "", err
63 }
64 > return base64.RawURLEncoding.EncodeToString(b), nil nexus_completion.go
65 }
66
go.temporal.io/server/chasm/lib/scheduler/scheduler.go 14 introduced LOC · 1 range

Open complete file

1055 func (s *Scheduler) startWorkflowSearchAttributes(
1056 nominal time.Time,
1057 > ) *commonpb.SearchAttributes { scheduler.go
1058 > scheduledStartTime := chasm.SearchAttributeTemporalScheduledStartTime.Value(nominal)
1059 > scheduledByID := chasm.SearchAttributeTemporalScheduledByID.Value(s.ScheduleId)
1060 > fields := payload.MergeMapOfPayload(
1061 > s.Schedule.GetAction().GetStartWorkflow().GetSearchAttributes().GetIndexedFields(),
1062 > map[string]*commonpb.Payload{
1063 > scheduledStartTime.Field: scheduledStartTime.Value.MustEncode(),
1064 > scheduledByID.Field: scheduledByID.Value.MustEncode(),
1065 > },
1066 > )
1067 > return &commonpb.SearchAttributes{
1068 > IndexedFields: fields,
1069 > }
1070 > }
1071
1072 func userCustomSearchAttributes(fields map[string]*commonpb.Payload) map[string]*commonpb.Payload {
go.temporal.io/server/common/testing/mockapi/workflowservicemock/v1/service_grpc.pb.mock.go 13 introduced LOC · 3 ranges

Open complete file

2044
2045 // StartWorkflowExecution mocks base method.
2046 > func (m *MockWorkflowServiceClient) StartWorkflowExecution(ctx context.Context, in *workflowservice.StartWorkflowExecutionRequest, opts ...grpc.CallOption) (*workflowservice.StartWorkflowExecutionResponse, error) { service_grpc.pb.mock.go
2047 > m.ctrl.T.Helper()
2048 > varargs := []any{ctx, in}
2049 > for _, a := range opts {
2050 varargs = append(varargs, a)
2051 }
2052 > ret := m.ctrl.Call(m, "StartWorkflowExecution", varargs...) service_grpc.pb.mock.go
2053 > ret0, _ := ret[0].(*workflowservice.StartWorkflowExecutionResponse)
2054 > ret1, _ := ret[1].(error)
2055 > return ret0, ret1
2056 }
2057
2058 // StartWorkflowExecution indicates an expected call of StartWorkflowExecution.
2059 > func (mr *MockWorkflowServiceClientMockRecorder) StartWorkflowExecution(ctx, in any, opts ...any) *gomock.Call { service_grpc.pb.mock.go
2060 > mr.mock.ctrl.T.Helper()
2061 > varargs := append([]any{ctx, in}, opts...)
2062 > return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "StartWorkflowExecution", reflect.TypeOf((*MockWorkflowServiceClient)(nil).StartWorkflowExecution), varargs...)
2063 > }
2064
2065 // StopBatchOperation mocks base method.

Direct hierarchy

Parents

Children