Vihren Atlas

request_response.pb.go ×6

Code frontier

unlabeled · c_a821276bda0e

2 tests · 45591 LOC · 755 files · introduces 0 tests · 246 LOC · 19 files

Introduces — evidence that enters the hierarchy at this concept

Code
50 ranges246 lines · 19 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
10778 ranges45591 lines · 755 files · Browse complete extent
All tests (intent)
2 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.

19 files ranked by introduced lines: 246 introduced LOC across 50 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/api/historyservice/v1/request_response.pb.go 30 introduced LOC · 6 ranges

Open complete file

1776 }
1777
1778 > func (x *RecordActivityTaskStartedRequest) Reset() { request_response.pb.go
1779 > *x = RecordActivityTaskStartedRequest{}
1780 > mi := &file_temporal_server_api_historyservice_v1_request_response_proto_msgTypes[14]
1781 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
1782 > ms.StoreMessageInfo(mi)
1783 > }
1784
1785 func (x *RecordActivityTaskStartedRequest) String() string {
1792 mi := &file_temporal_server_api_historyservice_v1_request_response_proto_msgTypes[14]
1793 if x != nil {
1794 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) request_response.pb.go
1795 > if ms.LoadMessageInfo() == nil {
1796 > ms.StoreMessageInfo(mi)
1797 > }
1798 > return ms
1799 }
1800 return mi.MessageOf(x)
1813 }
1814
1815 > func (x *RecordActivityTaskStartedRequest) GetWorkflowExecution() *v14.WorkflowExecution { request_response.pb.go
1816 > if x != nil {
1817 > return x.WorkflowExecution
1818 > }
1819 return nil
1820 }
1883 }
1884
1885 > func (x *RecordActivityTaskStartedRequest) GetComponentRef() []byte { request_response.pb.go
1886 > if x != nil {
1887 > return x.ComponentRef
1888 > }
1889 return nil
1890 }
1910 }
1911
1912 > func (x *RecordActivityTaskStartedResponse) Reset() { request_response.pb.go
1913 > *x = RecordActivityTaskStartedResponse{}
1914 > mi := &file_temporal_server_api_historyservice_v1_request_response_proto_msgTypes[15]
1915 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
1916 > ms.StoreMessageInfo(mi)
1917 > }
1918
1919 func (x *RecordActivityTaskStartedResponse) String() string {
1926 mi := &file_temporal_server_api_historyservice_v1_request_response_proto_msgTypes[15]
1927 if x != nil {
1928 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) request_response.pb.go
1929 > if ms.LoadMessageInfo() == nil {
1930 > ms.StoreMessageInfo(mi)
1931 > }
1932 > return ms
1933 }
1934 return mi.MessageOf(x)
go.temporal.io/server/api/matchingservice/v1/request_response.pb.go 24 introduced LOC · 4 ranges

Open complete file

1030 }
1031
1032 > func (x *AddActivityTaskRequest) Reset() { request_response.pb.go
1033 > *x = AddActivityTaskRequest{}
1034 > mi := &file_temporal_server_api_matchingservice_v1_request_response_proto_msgTypes[7]
1035 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
1036 > ms.StoreMessageInfo(mi)
1037 > }
1038
1039 func (x *AddActivityTaskRequest) String() string {
1074 }
1075
1076 > func (x *AddActivityTaskRequest) GetTaskQueue() *v14.TaskQueue { request_response.pb.go
1077 > if x != nil {
1078 > return x.TaskQueue
1079 > }
1080 return nil
1081 }
1146 }
1147
1148 > func (x *AddActivityTaskResponse) Reset() { request_response.pb.go
1149 > *x = AddActivityTaskResponse{}
1150 > mi := &file_temporal_server_api_matchingservice_v1_request_response_proto_msgTypes[8]
1151 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
1152 > ms.StoreMessageInfo(mi)
1153 > }
1154
1155 func (x *AddActivityTaskResponse) String() string {
1159 func (*AddActivityTaskResponse) ProtoMessage() {}
1160
1161 > func (x *AddActivityTaskResponse) ProtoReflect() protoreflect.Message { request_response.pb.go
1162 > mi := &file_temporal_server_api_matchingservice_v1_request_response_proto_msgTypes[8]
1163 > if x != nil {
1164 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
1165 > if ms.LoadMessageInfo() == nil {
1166 > ms.StoreMessageInfo(mi)
1167 > }
1168 > return ms
1169 }
1170 return mi.MessageOf(x)
go.temporal.io/server/client/matching/client.go 22 introduced LOC · 4 ranges

Open complete file

157 request *matchingservice.AddActivityTaskRequest,
158 opts ...grpc.CallOption,
159 > ) (*matchingservice.AddActivityTaskResponse, error) { client.go
160 > if !isPartitionAwareKind(request.GetTaskQueue().GetKind()) {
161 return c.addActivityTask(ctx, PartitionCounts{}, request, opts)
162 }
163 > pkey := c.partitionCache.makeKey( client.go
164 > request.GetNamespaceId(),
165 > request.GetTaskQueue().GetName(),
166 > enumspb.TASK_QUEUE_TYPE_ACTIVITY,
167 > )
168 > return invokeWithPartitionCounts(ctx, c.logger, c.partitionCache, pkey, request, opts, c.addActivityTask)
169 }
170
174 request *matchingservice.AddActivityTaskRequest,
175 opts []grpc.CallOption,
176 > ) (*matchingservice.AddActivityTaskResponse, error) { client.go
177 > request = common.CloneProto(request)
178 > client, err := c.pickClientForWrite(
179 > request.GetTaskQueue(),
180 > request.GetNamespaceId(),
181 > enumspb.TASK_QUEUE_TYPE_ACTIVITY,
182 > request.GetForwardInfo().GetSourcePartition(),
183 > pc,
184 > )
185 > if err != nil {
186 return nil, err
187 }
188 > ctx, cancel := c.createContext(ctx) client.go
189 > defer cancel()
190 >
191 > return client.AddActivityTask(ctx, request, opts...)
192 }
193
go.temporal.io/server/service/frontend/workflow_handler.go 22 introduced LOC · 1 range

Open complete file

1398 }
1399
1400 > return &workflowservice.PollActivityTaskQueueResponse{ workflow_handler.go
1401 > TaskToken: matchingResponse.TaskToken,
1402 > WorkflowExecution: matchingResponse.WorkflowExecution,
1403 > ActivityId: matchingResponse.ActivityId,
1404 > ActivityType: matchingResponse.ActivityType,
1405 > ActivityRunId: matchingResponse.ActivityRunId,
1406 > Input: matchingResponse.Input,
1407 > ScheduledTime: matchingResponse.ScheduledTime,
1408 > ScheduleToCloseTimeout: matchingResponse.ScheduleToCloseTimeout,
1409 > StartedTime: matchingResponse.StartedTime,
1410 > StartToCloseTimeout: matchingResponse.StartToCloseTimeout,
1411 > HeartbeatTimeout: matchingResponse.HeartbeatTimeout,
1412 > Attempt: matchingResponse.Attempt,
1413 > CurrentAttemptScheduledTime: matchingResponse.CurrentAttemptScheduledTime,
1414 > HeartbeatDetails: matchingResponse.HeartbeatDetails,
1415 > WorkflowType: matchingResponse.WorkflowType,
1416 > WorkflowNamespace: matchingResponse.WorkflowNamespace,
1417 > Header: matchingResponse.Header,
1418 > PollerScalingDecision: matchingResponse.PollerScalingDecision,
1419 > Priority: matchingResponse.Priority,
1420 > RetryPolicy: matchingResponse.RetryPolicy,
1421 > }, nil
1422 }
1423
go.temporal.io/server/service/matching/pri_forwarder.go 20 introduced LOC · 2 ranges

Open complete file

102 },
103 )
104 > case enumspb.TASK_QUEUE_TYPE_ACTIVITY: pri_forwarder.go
105 > _, err = f.client.AddActivityTask(
106 > ctx, &matchingservice.AddActivityTaskRequest{
107 > NamespaceId: task.event.Data.GetNamespaceId(),
108 > Execution: task.workflowExecution(),
109 > TaskQueue: &taskqueuepb.TaskQueue{
110 > Name: target.RpcName(),
111 > Kind: f.partition.Kind(),
112 > },
113 > ScheduledEventId: task.event.Data.GetScheduledEventId(),
114 > Clock: task.event.Data.GetClock(),
115 > ScheduleToStartTimeout: expirationDuration,
116 > ForwardInfo: f.getForwardInfo(task),
117 > VersionDirective: task.event.Data.GetVersionDirective(),
118 > Stamp: task.event.Data.GetStamp(),
119 > Priority: task.event.Data.GetPriority(),
120 > ComponentRef: task.event.Data.GetComponentRef(),
121 > },
122 > )
123 default:
124 return errInvalidTaskQueueType
275 return nil, errNoTasks
276 }
277 > return newInternalStartedTask(&startedTaskInfo{activityTaskInfo: resp}), nil pri_forwarder.go
278 case enumspb.TASK_QUEUE_TYPE_NEXUS:
279 resp, err := client.PollNexusTaskQueue(ctx, &matchingservice.PollNexusTaskQueueRequest{
go.temporal.io/server/client/history/client.go 18 introduced LOC · 3 ranges

Open complete file

222 request *historyservice.RecordActivityTaskStartedRequest,
223 opts ...grpc.CallOption,
224 > ) (*historyservice.RecordActivityTaskStartedResponse, error) { client.go
225 > var shardID int32
226 >
227 > // For Chasm components we need to route the shard based on business ID. Note that shardIDFromWorkflowID simply
228 > // calculates the hash from the ID so it works for both workflowID and businessID.
229 > if len(request.GetComponentRef()) == 0 {
230 > shardID = c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetWorkflowExecution().GetWorkflowId())
231 > } else {
232 componentRef, err := c.tokenSerializer.DeserializeChasmComponentRef(request.GetComponentRef())
233 if err != nil {
238 }
239
240 > var response *historyservice.RecordActivityTaskStartedResponse client.go
241 > op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
242 > var err error
243 > ctx, cancel := c.createContext(ctx)
244 > defer cancel()
245 > response, err = client.RecordActivityTaskStarted(ctx, request, opts...)
246 > return err
247 > }
248 > if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
249 return nil, err
250 }
251 > return response, nil client.go
252 }
253
go.temporal.io/server/service/matching/handler.go 18 introduced LOC · 2 ranges

Open complete file

177 ctx context.Context,
178 request *matchingservice.AddActivityTaskRequest,
179 > ) (_ *matchingservice.AddActivityTaskResponse, retError error) { handler.go
180 > defer log.CapturePanic(h.logger, &retError)
181 > startT := time.Now().UTC()
182 > opMetrics := h.opMetricsHandler(
183 > request.GetNamespaceId(),
184 > request.GetTaskQueue(),
185 > enumspb.TASK_QUEUE_TYPE_ACTIVITY,
186 > metrics.MatchingAddActivityTaskScope,
187 > )
188 >
189 > if request.GetForwardInfo() != nil {
190 > h.reportForwardedPerTaskQueueCounter(opMetrics, namespace.ID(request.GetNamespaceId()))
191 > }
192
193 > assignedBuildId, syncMatch, err := h.engine.AddActivityTask(ctx, request) handler.go
194 > if syncMatch {
195 > metrics.SyncMatchLatencyPerTaskQueue.With(opMetrics).Record(time.Since(startT))
196 > }
197 > return &matchingservice.AddActivityTaskResponse{AssignedBuildId: assignedBuildId}, err
198 }
199
go.temporal.io/server/api/historyservice/v1/service_grpc.pb.go 17 introduced LOC · 5 ranges

Open complete file

445 }
446
447 > func (c *historyServiceClient) RecordActivityTaskStarted(ctx context.Context, in *RecordActivityTaskStartedRequest, opts ...grpc.CallOption) (*RecordActivityTaskStartedResponse, error) { service_grpc.pb.go
448 > out := new(RecordActivityTaskStartedResponse)
449 > err := c.cc.Invoke(ctx, HistoryService_RecordActivityTaskStarted_FullMethodName, in, out, opts...)
450 > if err != nil {
451 return nil, err
452 }
453 > return out, nil service_grpc.pb.go
454 }
455
1746 }
1747
1748 > func _HistoryService_RecordActivityTaskStarted_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { service_grpc.pb.go
1749 > in := new(RecordActivityTaskStartedRequest)
1750 > if err := dec(in); err != nil {
1751 return nil, err
1752 }
1753 > if interceptor == nil { service_grpc.pb.go
1754 return srv.(HistoryServiceServer).RecordActivityTaskStarted(ctx, in)
1755 }
1756 > info := &grpc.UnaryServerInfo{ service_grpc.pb.go
1757 > Server: srv,
1758 > FullMethod: HistoryService_RecordActivityTaskStarted_FullMethodName,
1759 > }
1760 > handler := func(ctx context.Context, req interface{}) (interface{}, error) {
1761 > return srv.(HistoryServiceServer).RecordActivityTaskStarted(ctx, req.(*RecordActivityTaskStartedRequest))
1762 > }
1763 > return interceptor(ctx, in, info, handler)
1764 }
1765
go.temporal.io/server/api/matchingservice/v1/service_grpc.pb.go 17 introduced LOC · 5 ranges

Open complete file

290 }
291
292 > func (c *matchingServiceClient) AddActivityTask(ctx context.Context, in *AddActivityTaskRequest, opts ...grpc.CallOption) (*AddActivityTaskResponse, error) { service_grpc.pb.go
293 > out := new(AddActivityTaskResponse)
294 > err := c.cc.Invoke(ctx, MatchingService_AddActivityTask_FullMethodName, in, out, opts...)
295 > if err != nil {
296 return nil, err
297 }
298 > return out, nil service_grpc.pb.go
299 }
300
1029 }
1030
1031 > func _MatchingService_AddActivityTask_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { service_grpc.pb.go
1032 > in := new(AddActivityTaskRequest)
1033 > if err := dec(in); err != nil {
1034 return nil, err
1035 }
1036 > if interceptor == nil { service_grpc.pb.go
1037 return srv.(MatchingServiceServer).AddActivityTask(ctx, in)
1038 }
1039 > info := &grpc.UnaryServerInfo{ service_grpc.pb.go
1040 > Server: srv,
1041 > FullMethod: MatchingService_AddActivityTask_FullMethodName,
1042 > }
1043 > handler := func(ctx context.Context, req interface{}) (interface{}, error) {
1044 > return srv.(MatchingServiceServer).AddActivityTask(ctx, req.(*AddActivityTaskRequest))
1045 > }
1046 > return interceptor(ctx, in, info, handler)
1047 }
1048
go.temporal.io/server/client/matching/metric_client.go 13 introduced LOC · 2 ranges

Open complete file

46 request *matchingservice.AddActivityTaskRequest,
47 opts ...grpc.CallOption,
48 > ) (_ *matchingservice.AddActivityTaskResponse, retError error) { metric_client.go
49 >
50 > scope, stopwatch := c.startMetricsRecording(ctx, metrics.MatchingClientAddActivityTaskScope)
51 > defer func() {
52 > c.finishMetricsRecording(scope, stopwatch, retError)
53 > }()
54
55 > c.emitForwardedSourceStats( metric_client.go
56 > scope,
57 > request.GetForwardInfo().GetSourcePartition(),
58 > request.TaskQueue,
59 > )
60 >
61 > return c.client.AddActivityTask(ctx, request, opts...)
62 }
63
go.temporal.io/server/service/history/handler.go 12 introduced LOC · 6 ranges

Open complete file

318
319 // RecordActivityTaskStarted - Record Activity Task started.
320 > func (h *Handler) RecordActivityTaskStarted(ctx context.Context, request *historyservice.RecordActivityTaskStartedRequest) (*historyservice.RecordActivityTaskStartedResponse, error) { handler.go
321 > // Handle as standalone activity if request has component ref.
322 > if activityRefProto := request.GetComponentRef(); len(activityRefProto) > 0 {
323 response, _, err := chasm.UpdateComponent(
324 ctx,
334
335 // Handle worklow activity (mutable state backed implementation).
336 > namespaceID := namespace.ID(request.GetNamespaceId()) handler.go
337 > if namespaceID == "" {
338 return nil, h.convertError(errNamespaceNotSet)
339 }
340
341 > shardContext, err := h.controller.GetShardByNamespaceWorkflow(namespaceID, request.GetWorkflowExecution().GetWorkflowId()) handler.go
342 > if err != nil {
343 return nil, h.convertError(err)
344 }
345 > engine, err := shardContext.GetEngine(ctx) handler.go
346 > if err != nil {
347 return nil, h.convertError(err)
348 }
349
350 > response, err := engine.RecordActivityTaskStarted(ctx, request) handler.go
351 > if err != nil {
352 return nil, h.convertError(err)
353 }
354 > return response, nil handler.go
355 }
356
go.temporal.io/server/client/history/retryable_client_gen.go 9 introduced LOC · 1 range

Open complete file

661 request *historyservice.RecordActivityTaskStartedRequest,
662 opts ...grpc.CallOption,
663 > ) (*historyservice.RecordActivityTaskStartedResponse, error) { retryable_client_gen.go
664 > var resp *historyservice.RecordActivityTaskStartedResponse
665 > op := func(ctx context.Context) error {
666 > var err error
667 > resp, err = c.client.RecordActivityTaskStarted(ctx, request, opts...)
668 > return err
669 > }
670 > err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
671 > return resp, err
672 }
673
go.temporal.io/server/client/history/metric_client_gen.go 7 introduced LOC · 2 ranges

Open complete file

602 request *historyservice.RecordActivityTaskStartedRequest,
603 opts ...grpc.CallOption,
604 > ) (_ *historyservice.RecordActivityTaskStartedResponse, retError error) { metric_client_gen.go
605 >
606 > metricsHandler, startTime := c.startMetricsRecording(ctx, "HistoryClientRecordActivityTaskStarted")
607 > defer func() {
608 > c.finishMetricsRecording(metricsHandler, startTime, retError)
609 > }()
610
611 > return c.client.RecordActivityTaskStarted(ctx, request, opts...) metric_client_gen.go
612 }
613
go.temporal.io/server/service/history/queues/scheduler.go 5 introduced LOC · 1 range

Open complete file

104
105 taskChannelKeyFn := func(e Executable) TaskChannelKey {
106 > return TaskChannelKey{ scheduler.go
107 > NamespaceID: e.GetNamespaceID(),
108 > Priority: e.GetPriority(),
109 > }
110 > }
111 channelWeightFn := func(key TaskChannelKey) int {
112 namespaceWeights := options.ActiveNamespaceWeights
go.temporal.io/server/service/history/queues/executable.go 3 introduced LOC · 1 range

Open complete file

758 }
759
760 > func (e *executableImpl) GetPriority() ctasks.Priority { executable.go
761 > return e.priority
762 > }
763
764 func (e *executableImpl) GetTask() tasks.Task {
go.temporal.io/server/service/history/transfer_queue_task_executor_base.go 3 introduced LOC · 1 range

Open complete file

129
130 if directive.GetUseAssignmentRules() == nil {
131 > // activity is not getting a new build ID, so no need to update MS transfer_queue_task_executor_base.go
132 > return nil
133 > }
134
135 return updateIndependentActivityBuildId(
go.temporal.io/server/service/matching/matching_engine.go 3 introduced LOC · 1 range

Open complete file

997
998 if task.isStarted() {
999 > // tasks received from remote are already started. So, simply forward the response matching_engine.go
1000 > return task.pollActivityTaskQueueResponse(), nil
1001 > }
1002 requestClone := request
1003 if versionSetUsed {
go.temporal.io/server/service/history/worker_versioning_util.go 2 introduced LOC · 2 ranges

Open complete file

229 } else if id := mutableState.GetAssignedBuildId(); id != "" {
230 return worker_versioning.MakeBuildIdDirective(id)
231 > } else if id := worker_versioning.BuildIdIfUsingVersioning(mutableState.GetMostRecentWorkerVersionStamp()); id != "" { worker_versioning_util.go
232 // TODO: old versioning only [cleanup-old-wv]
233 return worker_versioning.MakeBuildIdDirective(id)
234 }
235 // else: unversioned execution
236 > return nil worker_versioning_util.go
237 }
go.temporal.io/server/common/rpc/interceptor/telemetry.go 1 introduced LOC · 1 range

Open complete file

343 return
344 }
345 > if activityPollResponse.Attempt > 1 { telemetry.go
346 metrics.ActionCounter.With(metricsHandler).Record(1, metrics.ActionType("activity_retry"))
347 }

Direct hierarchy

Parents

Children