Vihren Atlas

activities.go ×12

Code frontier

unlabeled · c_60fb46d16948

3 tests · 2394 LOC · 116 files · introduces 0 tests · 65 LOC · 2 files

Introduces — evidence that enters the hierarchy at this concept

Code
15 ranges65 lines · 2 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
380 ranges2394 lines · 116 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: 65 introduced LOC across 15 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/service/worker/batcher/activities.go 52 introduced LOC · 12 ranges

Open complete file

583 metricsHandler metrics.Handler,
584 logger log.Logger,
585 > ) { activities.go
586 > for {
587 > select {
588 > case <-ctx.Done():
589 > return
590 > case task := <-taskCh:
591 > if isDone(ctx) {
592 return
593 }
594
595 > if task.executionInfo == nil && task.targetExecution == nil { activities.go
596 continue
597 }
598
599 > a.processTaskWithRetries(ctx, batchOperation, namespace, task, respCh, limiter, sdkClient, frontendClient, metricsHandler, logger) activities.go
600 }
601 }
605 // the identifying fields of the target of a single call. Deriving the request ID
606 // lets the server's idempotency checks collapse retries of the same call.
607 > func deterministicRequestID(jobID string, parts ...string) string { activities.go
608 > var key strings.Builder
609 > for _, part := range parts {
610 > key.WriteString(":" + part)
611 > }
612 > return uuid.NewSHA1(uuid.Nil, []byte(key.String())).String()
613 }
614
625 frontendClient workflowservice.WorkflowServiceClient,
626 logger log.Logger,
627 > ) error { activities.go
628 > var err error
629 >
630 > // Bound the processing of each individual task so a single hung operation
631 > // cannot block the processor indefinitely.
632 > ctx, cancel := taskTimeoutContext(ctx)
633 > defer cancel()
634 >
635 > // Handle admin batch operations
636 > if batchOperation.AdminRequest != nil {
637 return a.processAdminTask(ctx, batchOperation, task, limiter)
638 }
639
640 > switch operation := batchOperation.Request.Operation.(type) { activities.go
641 case *workflowservice.StartBatchOperationRequest_TerminateActivitiesOperation:
642 err = processTargetTask(ctx, limiter, task,
685 return sdkClient.CancelWorkflow(ctx, executionInfo.Execution.WorkflowId, executionInfo.Execution.RunId)
686 })
687 > case *workflowservice.StartBatchOperationRequest_SignalOperation: activities.go
688 > err = processTask(ctx, limiter, task,
689 > func(executionInfo *workflowpb.WorkflowExecutionInfo) error {
690 > _, err := frontendClient.SignalWorkflowExecution(ctx, &workflowservice.SignalWorkflowExecutionRequest{
691 > Namespace: namespace,
692 > WorkflowExecution: executionInfo.Execution,
693 > SignalName: operation.SignalOperation.GetSignal(),
694 > Input: operation.SignalOperation.GetInput(),
695 > Identity: operation.SignalOperation.GetIdentity(),
696 > RequestId: deterministicRequestID(batchOperation.Request.GetJobId(), "signal",
697 > executionInfo.Execution.GetWorkflowId(), executionInfo.Execution.GetRunId(), operation.SignalOperation.GetSignal()),
698 > })
699 > return err
700 > })
701 case *workflowservice.StartBatchOperationRequest_DeletionOperation:
702 err = processTask(ctx, limiter, task,
840 err = fmt.Errorf("unknown batch type: %v", batchOperation.BatchType)
841 }
842 > return err activities.go
843 }
844
875 metricsHandler metrics.Handler,
876 logger log.Logger,
877 > ) { activities.go
878 > var err error
879 > for {
880 > err = a.processSingleTask(ctx, batchOperation, ns, task, limiter, sdkClient, frontendClient, logger)
881 > if err == nil {
882 metrics.BatcherProcessorSuccess.With(metricsHandler).Record(1)
883 break
895
896 // Send one response per task; stop early if the context is cancelled.
897 > select { activities.go
898 > case respCh <- taskResponse{err: err, page: task.page}:
899 case <-ctx.Done():
900 }
1001 }
1002
1003 > func isDone(ctx context.Context) bool { activities.go
1004 > select {
1005 case <-ctx.Done():
1006 return true
1007 > default: activities.go
1008 > return false
1009 }
1010 }
go.temporal.io/server/common/testing/mockapi/workflowservicemock/v1/service_grpc.pb.mock.go 13 introduced LOC · 3 ranges

Open complete file

1964
1965 // SignalWorkflowExecution mocks base method.
1966 > func (m *MockWorkflowServiceClient) SignalWorkflowExecution(ctx context.Context, in *workflowservice.SignalWorkflowExecutionRequest, opts ...grpc.CallOption) (*workflowservice.SignalWorkflowExecutionResponse, error) { service_grpc.pb.mock.go
1967 > m.ctrl.T.Helper()
1968 > varargs := []any{ctx, in}
1969 > for _, a := range opts {
1970 varargs = append(varargs, a)
1971 }
1972 > ret := m.ctrl.Call(m, "SignalWorkflowExecution", varargs...) service_grpc.pb.mock.go
1973 > ret0, _ := ret[0].(*workflowservice.SignalWorkflowExecutionResponse)
1974 > ret1, _ := ret[1].(error)
1975 > return ret0, ret1
1976 }
1977
1978 // SignalWorkflowExecution indicates an expected call of SignalWorkflowExecution.
1979 > func (mr *MockWorkflowServiceClientMockRecorder) SignalWorkflowExecution(ctx, in any, opts ...any) *gomock.Call { service_grpc.pb.mock.go
1980 > mr.mock.ctrl.T.Helper()
1981 > varargs := append([]any{ctx, in}, opts...)
1982 > return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SignalWorkflowExecution", reflect.TypeOf((*MockWorkflowServiceClient)(nil).SignalWorkflowExecution), varargs...)
1983 > }
1984
1985 // StartActivityExecution mocks base method.

Direct hierarchy

Parents

Children