Vihren Atlas

service_grpc.pb.go ×19

Code frontier

unlabeled · c_fe71aaab2847

10 tests · 25769 LOC · 622 files · introduces 0 tests · 357 LOC · 19 files

Introduces — evidence that enters the hierarchy at this concept

Code
80 ranges357 lines · 19 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
5314 ranges25769 lines · 622 files · Browse complete extent
All tests (intent)
10 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: 357 introduced LOC across 80 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/api/matchingservice/v1/request_response.pb.go 104 introduced LOC · 17 ranges

Open complete file

1435 }
1436
1437 > func (x *CancelOutstandingPollRequest) Reset() { request_response.pb.go
1438 > *x = CancelOutstandingPollRequest{}
1439 > mi := &file_temporal_server_api_matchingservice_v1_request_response_proto_msgTypes[13]
1440 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
1441 > ms.StoreMessageInfo(mi)
1442 > }
1443
1444 func (x *CancelOutstandingPollRequest) String() string {
1448 func (*CancelOutstandingPollRequest) ProtoMessage() {}
1449
1450 > func (x *CancelOutstandingPollRequest) ProtoReflect() protoreflect.Message { request_response.pb.go
1451 > mi := &file_temporal_server_api_matchingservice_v1_request_response_proto_msgTypes[13]
1452 > if x != nil {
1453 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
1454 > if ms.LoadMessageInfo() == nil {
1455 > ms.StoreMessageInfo(mi)
1456 > }
1457 > return ms
1458 }
1459 return mi.MessageOf(x)
1465 }
1466
1467 > func (x *CancelOutstandingPollRequest) GetNamespaceId() string { request_response.pb.go
1468 > if x != nil {
1469 > return x.NamespaceId
1470 > }
1471 return ""
1472 }
1473
1474 > func (x *CancelOutstandingPollRequest) GetTaskQueueType() v19.TaskQueueType { request_response.pb.go
1475 > if x != nil {
1476 > return x.TaskQueueType
1477 > }
1478 return v19.TaskQueueType(0)
1479 }
1480
1481 > func (x *CancelOutstandingPollRequest) GetTaskQueue() *v14.TaskQueue { request_response.pb.go
1482 > if x != nil {
1483 > return x.TaskQueue
1484 > }
1485 return nil
1486 }
1499 }
1500
1501 > func (x *CancelOutstandingPollResponse) Reset() { request_response.pb.go
1502 > *x = CancelOutstandingPollResponse{}
1503 > mi := &file_temporal_server_api_matchingservice_v1_request_response_proto_msgTypes[14]
1504 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
1505 > ms.StoreMessageInfo(mi)
1506 > }
1507
1508 func (x *CancelOutstandingPollResponse) String() string {
1512 func (*CancelOutstandingPollResponse) ProtoMessage() {}
1513
1514 > func (x *CancelOutstandingPollResponse) ProtoReflect() protoreflect.Message { request_response.pb.go
1515 > mi := &file_temporal_server_api_matchingservice_v1_request_response_proto_msgTypes[14]
1516 > if x != nil {
1517 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
1518 > if ms.LoadMessageInfo() == nil {
1519 > ms.StoreMessageInfo(mi)
1520 > }
1521 > return ms
1522 }
1523 return mi.MessageOf(x)
3539 }
3540
3541 > func (x *ForceUnloadTaskQueuePartitionRequest) Reset() { request_response.pb.go
3542 > *x = ForceUnloadTaskQueuePartitionRequest{}
3543 > mi := &file_temporal_server_api_matchingservice_v1_request_response_proto_msgTypes[47]
3544 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
3545 > ms.StoreMessageInfo(mi)
3546 > }
3547
3548 func (x *ForceUnloadTaskQueuePartitionRequest) String() string {
3552 func (*ForceUnloadTaskQueuePartitionRequest) ProtoMessage() {}
3553
3554 > func (x *ForceUnloadTaskQueuePartitionRequest) ProtoReflect() protoreflect.Message { request_response.pb.go
3555 > mi := &file_temporal_server_api_matchingservice_v1_request_response_proto_msgTypes[47]
3556 > if x != nil {
3557 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
3558 > if ms.LoadMessageInfo() == nil {
3559 > ms.StoreMessageInfo(mi)
3560 > }
3561 > return ms
3562 }
3563 return mi.MessageOf(x)
3590 }
3591
3592 > func (x *ForceUnloadTaskQueuePartitionResponse) Reset() { request_response.pb.go
3593 > *x = ForceUnloadTaskQueuePartitionResponse{}
3594 > mi := &file_temporal_server_api_matchingservice_v1_request_response_proto_msgTypes[48]
3595 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
3596 > ms.StoreMessageInfo(mi)
3597 > }
3598
3599 func (x *ForceUnloadTaskQueuePartitionResponse) String() string {
3603 func (*ForceUnloadTaskQueuePartitionResponse) ProtoMessage() {}
3604
3605 > func (x *ForceUnloadTaskQueuePartitionResponse) ProtoReflect() protoreflect.Message { request_response.pb.go
3606 > mi := &file_temporal_server_api_matchingservice_v1_request_response_proto_msgTypes[48]
3607 > if x != nil {
3608 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
3609 > if ms.LoadMessageInfo() == nil {
3610 > ms.StoreMessageInfo(mi)
3611 > }
3612 > return ms
3613 }
3614 return mi.MessageOf(x)
4917 }
4918
4919 > func (x *RecordWorkerHeartbeatRequest) Reset() { request_response.pb.go
4920 > *x = RecordWorkerHeartbeatRequest{}
4921 > mi := &file_temporal_server_api_matchingservice_v1_request_response_proto_msgTypes[71]
4922 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
4923 > ms.StoreMessageInfo(mi)
4924 > }
4925
4926 func (x *RecordWorkerHeartbeatRequest) String() string {
4930 func (*RecordWorkerHeartbeatRequest) ProtoMessage() {}
4931
4932 > func (x *RecordWorkerHeartbeatRequest) ProtoReflect() protoreflect.Message { request_response.pb.go
4933 > mi := &file_temporal_server_api_matchingservice_v1_request_response_proto_msgTypes[71]
4934 > if x != nil {
4935 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
4936 > if ms.LoadMessageInfo() == nil {
4937 > ms.StoreMessageInfo(mi)
4938 > }
4939 > return ms
4940 }
4941 return mi.MessageOf(x)
4947 }
4948
4949 > func (x *RecordWorkerHeartbeatRequest) GetNamespaceId() string { request_response.pb.go
4950 > if x != nil {
4951 > return x.NamespaceId
4952 > }
4953 return ""
4954 }
4955
4956 > func (x *RecordWorkerHeartbeatRequest) GetHeartbeartRequest() *v1.RecordWorkerHeartbeatRequest { request_response.pb.go
4957 > if x != nil {
4958 > return x.HeartbeartRequest
4959 > }
4960 return nil
4961 }
4967 }
4968
4969 > func (x *RecordWorkerHeartbeatResponse) Reset() { request_response.pb.go
4970 > *x = RecordWorkerHeartbeatResponse{}
4971 > mi := &file_temporal_server_api_matchingservice_v1_request_response_proto_msgTypes[72]
4972 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
4973 > ms.StoreMessageInfo(mi)
4974 > }
4975
4976 func (x *RecordWorkerHeartbeatResponse) String() string {
4980 func (*RecordWorkerHeartbeatResponse) ProtoMessage() {}
4981
4982 > func (x *RecordWorkerHeartbeatResponse) ProtoReflect() protoreflect.Message { request_response.pb.go
4983 > mi := &file_temporal_server_api_matchingservice_v1_request_response_proto_msgTypes[72]
4984 > if x != nil {
4985 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
4986 > if ms.LoadMessageInfo() == nil {
4987 > ms.StoreMessageInfo(mi)
4988 > }
4989 > return ms
4990 }
4991 return mi.MessageOf(x)
go.temporal.io/server/api/matchingservice/v1/service_grpc.pb.go 59 introduced LOC · 19 ranges

Open complete file

267 err := c.cc.Invoke(ctx, MatchingService_PollWorkflowTaskQueue_FullMethodName, in, out, opts...)
268 if err != nil {
269 > return nil, err service_grpc.pb.go
270 > }
271 return out, nil
272 }
276 err := c.cc.Invoke(ctx, MatchingService_PollActivityTaskQueue_FullMethodName, in, out, opts...)
277 if err != nil {
278 > return nil, err service_grpc.pb.go
279 > }
280 return out, nil
281 }
353 }
354
355 > func (c *matchingServiceClient) CancelOutstandingPoll(ctx context.Context, in *CancelOutstandingPollRequest, opts ...grpc.CallOption) (*CancelOutstandingPollResponse, error) { service_grpc.pb.go
356 > out := new(CancelOutstandingPollResponse)
357 > err := c.cc.Invoke(ctx, MatchingService_CancelOutstandingPoll_FullMethodName, in, out, opts...)
358 > if err != nil {
359 return nil, err
360 }
361 > return out, nil service_grpc.pb.go
362 }
363
438 err := c.cc.Invoke(ctx, MatchingService_GetTaskQueueUserData_FullMethodName, in, out, opts...)
439 if err != nil {
440 > return nil, err service_grpc.pb.go
441 > }
442 return out, nil
443 }
506 }
507
508 > func (c *matchingServiceClient) ForceUnloadTaskQueuePartition(ctx context.Context, in *ForceUnloadTaskQueuePartitionRequest, opts ...grpc.CallOption) (*ForceUnloadTaskQueuePartitionResponse, error) { service_grpc.pb.go
509 > out := new(ForceUnloadTaskQueuePartitionResponse)
510 > err := c.cc.Invoke(ctx, MatchingService_ForceUnloadTaskQueuePartition_FullMethodName, in, out, opts...)
511 > if err != nil {
512 return nil, err
513 }
514 > return out, nil service_grpc.pb.go
515 }
516
573 err := c.cc.Invoke(ctx, MatchingService_ListNexusEndpoints_FullMethodName, in, out, opts...)
574 if err != nil {
575 > return nil, err service_grpc.pb.go
576 > }
577 return out, nil
578 }
579
580 > func (c *matchingServiceClient) RecordWorkerHeartbeat(ctx context.Context, in *RecordWorkerHeartbeatRequest, opts ...grpc.CallOption) (*RecordWorkerHeartbeatResponse, error) { service_grpc.pb.go
581 > out := new(RecordWorkerHeartbeatResponse)
582 > err := c.cc.Invoke(ctx, MatchingService_RecordWorkerHeartbeat_FullMethodName, in, out, opts...)
583 > if err != nil {
584 return nil, err
585 }
586 > return out, nil service_grpc.pb.go
587 }
588
1155 }
1156
1157 > func _MatchingService_CancelOutstandingPoll_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { service_grpc.pb.go
1158 > in := new(CancelOutstandingPollRequest)
1159 > if err := dec(in); err != nil {
1160 return nil, err
1161 }
1162 > if interceptor == nil { service_grpc.pb.go
1163 return srv.(MatchingServiceServer).CancelOutstandingPoll(ctx, in)
1164 }
1165 > info := &grpc.UnaryServerInfo{ service_grpc.pb.go
1166 > Server: srv,
1167 > FullMethod: MatchingService_CancelOutstandingPoll_FullMethodName,
1168 > }
1169 > handler := func(ctx context.Context, req interface{}) (interface{}, error) {
1170 > return srv.(MatchingServiceServer).CancelOutstandingPoll(ctx, req.(*CancelOutstandingPollRequest))
1171 > }
1172 > return interceptor(ctx, in, info, handler)
1173 }
1174
1461 }
1462
1463 > func _MatchingService_ForceUnloadTaskQueuePartition_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { service_grpc.pb.go
1464 > in := new(ForceUnloadTaskQueuePartitionRequest)
1465 > if err := dec(in); err != nil {
1466 return nil, err
1467 }
1468 > if interceptor == nil { service_grpc.pb.go
1469 return srv.(MatchingServiceServer).ForceUnloadTaskQueuePartition(ctx, in)
1470 }
1471 > info := &grpc.UnaryServerInfo{ service_grpc.pb.go
1472 > Server: srv,
1473 > FullMethod: MatchingService_ForceUnloadTaskQueuePartition_FullMethodName,
1474 > }
1475 > handler := func(ctx context.Context, req interface{}) (interface{}, error) {
1476 > return srv.(MatchingServiceServer).ForceUnloadTaskQueuePartition(ctx, req.(*ForceUnloadTaskQueuePartitionRequest))
1477 > }
1478 > return interceptor(ctx, in, info, handler)
1479 }
1480
1605 }
1606
1607 > func _MatchingService_RecordWorkerHeartbeat_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { service_grpc.pb.go
1608 > in := new(RecordWorkerHeartbeatRequest)
1609 > if err := dec(in); err != nil {
1610 return nil, err
1611 }
1612 > if interceptor == nil { service_grpc.pb.go
1613 return srv.(MatchingServiceServer).RecordWorkerHeartbeat(ctx, in)
1614 }
1615 > info := &grpc.UnaryServerInfo{ service_grpc.pb.go
1616 > Server: srv,
1617 > FullMethod: MatchingService_RecordWorkerHeartbeat_FullMethodName,
1618 > }
1619 > handler := func(ctx context.Context, req interface{}) (interface{}, error) {
1620 > return srv.(MatchingServiceServer).RecordWorkerHeartbeat(ctx, req.(*RecordWorkerHeartbeatRequest))
1621 > }
1622 > return interceptor(ctx, in, info, handler)
1623 }
1624
go.temporal.io/server/service/frontend/workflow_handler.go 37 introduced LOC · 6 ranges

Open complete file

1116 })
1117 if err != nil {
1118 > contextWasCanceled := wh.cancelOutstandingPoll(childCtx, namespaceID, enumspb.TASK_QUEUE_TYPE_WORKFLOW, request.TaskQueue, pollerID) workflow_handler.go
1119 > if contextWasCanceled {
1120 > // Clear error as we don't want to report context cancellation error to count against our SLA.
1121 > // It doesn't matter what to return here, client has already gone. But (nil,nil) is invalid gogo return pair.
1122 > return &workflowservice.PollWorkflowTaskQueueResponse{}, nil
1123 > }
1124
1125 // These errors are expected from some versioning situations. We should not log them, it'd be too noisy.
1370 })
1371 if err != nil {
1372 > contextWasCanceled := wh.cancelOutstandingPoll(childCtx, namespaceID, enumspb.TASK_QUEUE_TYPE_ACTIVITY, request.TaskQueue, pollerID) workflow_handler.go
1373 > if contextWasCanceled {
1374 > // Clear error as we don't want to report context cancellation error to count against our SLA.
1375 > // It doesn't matter what to return here, client has already gone. But (nil,nil) is invalid gogo return pair.
1376 > return &workflowservice.PollActivityTaskQueueResponse{}, nil
1377 > }
1378
1379 // These errors are expected from some versioning situations. We should not log them, it'd be too noisy.
3023 // Record final heartbeat (best-effort)
3024 if request.WorkerHeartbeat != nil && wh.config.WorkerHeartbeatsEnabled(request.GetNamespace()) {
3025 > waitGroup.Go(func() { workflow_handler.go
3026 > _, err := wh.matchingClient.RecordWorkerHeartbeat(ctx, &matchingservice.RecordWorkerHeartbeatRequest{
3027 > NamespaceId: namespaceID.String(),
3028 > HeartbeartRequest: &workflowservice.RecordWorkerHeartbeatRequest{
3029 > Namespace: request.Namespace,
3030 > Identity: request.Identity,
3031 > WorkerHeartbeat: []*workerpb.WorkerHeartbeat{request.WorkerHeartbeat},
3032 > },
3033 > })
3034 > if err != nil {
3035 wh.logger.Error("Failed to record worker heartbeat during shutdown.",
3036 tag.WorkflowTaskQueueName(request.WorkerHeartbeat.GetTaskQueue()),
6708 taskQueue *taskqueuepb.TaskQueue,
6709 pollerID string,
6710 > ) bool { workflow_handler.go
6711 > // First check if this err is due to context cancellation. This means client connection to frontend is closed.
6712 > if !errors.Is(ctx.Err(), context.Canceled) {
6713 return false
6714 }
6716 // call to matching to notify this poller is gone to prevent any tasks being dispatched to zombie pollers.
6717 // TODO: specify a reasonable timeout for CancelOutstandingPoll.
6718 > _, err := wh.matchingClient.CancelOutstandingPoll( workflow_handler.go
6719 > rpc.CopyContextValues(context.TODO(), ctx),
6720 > &matchingservice.CancelOutstandingPollRequest{
6721 > NamespaceId: namespaceID.String(),
6722 > TaskQueueType: taskQueueType,
6723 > TaskQueue: taskQueue,
6724 > PollerId: pollerID,
6725 > },
6726 > )
6727 > // We can not do much if this call fails. Just log the error and move on.
6728 > if err != nil {
6729 wh.logger.Warn("Failed to cancel outstanding poller.",
6730 tag.WorkflowTaskQueueName(taskQueue.GetName()), tag.Error(err))
6731 }
6732
6733 > return true workflow_handler.go
6734 }
6735
go.temporal.io/server/client/matching/client_gen.go 27 introduced LOC · 8 ranges

Open complete file

38 request *matchingservice.CancelOutstandingPollRequest,
39 opts ...grpc.CallOption,
40 > ) (*matchingservice.CancelOutstandingPollResponse, error) { client_gen.go
41 >
42 > p, err := tqid.PartitionFromProto(request.GetTaskQueue(), request.GetNamespaceId(), request.GetTaskQueueType())
43 > if err != nil {
44 return nil, err
45 }
46
47 > client, err := c.getClientForTaskQueuePartition(p) client_gen.go
48 > if err != nil {
49 return nil, err
50 }
51 > ctx, cancel := c.createContext(ctx) client_gen.go
52 > defer cancel()
53 > return client.CancelOutstandingPoll(ctx, request, opts...)
54 }
55
309 request *matchingservice.ForceUnloadTaskQueuePartitionRequest,
310 opts ...grpc.CallOption,
311 > ) (*matchingservice.ForceUnloadTaskQueuePartitionResponse, error) { client_gen.go
312 >
313 > p := tqid.PartitionFromPartitionProto(request.GetTaskQueuePartition(), request.GetNamespaceId())
314 >
315 > client, err := c.getClientForTaskQueuePartition(p)
316 > if err != nil {
317 return nil, err
318 }
319 > ctx, cancel := c.createContext(ctx) client_gen.go
320 > defer cancel()
321 > return client.ForceUnloadTaskQueuePartition(ctx, request, opts...)
322 }
323
466 request *matchingservice.RecordWorkerHeartbeatRequest,
467 opts ...grpc.CallOption,
468 > ) (*matchingservice.RecordWorkerHeartbeatResponse, error) { client_gen.go
469 >
470 > p, err := tqid.NormalPartitionFromRpcName("not-applicable", request.GetNamespaceId(), enumspb.TASK_QUEUE_TYPE_UNSPECIFIED)
471 > if err != nil {
472 return nil, err
473 }
474
475 > client, err := c.getClientForTaskQueuePartition(p) client_gen.go
476 > if err != nil {
477 return nil, err
478 }
479 > ctx, cancel := c.createContext(ctx) client_gen.go
480 > defer cancel()
481 > return client.RecordWorkerHeartbeat(ctx, request, opts...)
482 }
483
go.temporal.io/server/client/matching/retryable_client_gen.go 27 introduced LOC · 3 ranges

Open complete file

61 request *matchingservice.CancelOutstandingPollRequest,
62 opts ...grpc.CallOption,
63 > ) (*matchingservice.CancelOutstandingPollResponse, error) { retryable_client_gen.go
64 > var resp *matchingservice.CancelOutstandingPollResponse
65 > op := func(ctx context.Context) error {
66 > var err error
67 > resp, err = c.client.CancelOutstandingPoll(ctx, request, opts...)
68 > return err
69 > }
70 > err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
71 > return resp, err
72 }
73
286 request *matchingservice.ForceUnloadTaskQueuePartitionRequest,
287 opts ...grpc.CallOption,
288 > ) (*matchingservice.ForceUnloadTaskQueuePartitionResponse, error) { retryable_client_gen.go
289 > var resp *matchingservice.ForceUnloadTaskQueuePartitionResponse
290 > op := func(ctx context.Context) error {
291 > var err error
292 > resp, err = c.client.ForceUnloadTaskQueuePartition(ctx, request, opts...)
293 > return err
294 > }
295 > err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
296 > return resp, err
297 }
298
466 request *matchingservice.RecordWorkerHeartbeatRequest,
467 opts ...grpc.CallOption,
468 > ) (*matchingservice.RecordWorkerHeartbeatResponse, error) { retryable_client_gen.go
469 > var resp *matchingservice.RecordWorkerHeartbeatResponse
470 > op := func(ctx context.Context) error {
471 > var err error
472 > resp, err = c.client.RecordWorkerHeartbeat(ctx, request, opts...)
473 > return err
474 > }
475 > err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
476 > return resp, err
477 }
478
go.temporal.io/server/client/matching/metric_client_gen.go 21 introduced LOC · 6 ranges

Open complete file

28 request *matchingservice.CancelOutstandingPollRequest,
29 opts ...grpc.CallOption,
30 > ) (_ *matchingservice.CancelOutstandingPollResponse, retError error) { metric_client_gen.go
31 >
32 > metricsHandler, startTime := c.startMetricsRecording(ctx, "MatchingClientCancelOutstandingPoll")
33 > defer func() {
34 > c.finishMetricsRecording(metricsHandler, startTime, retError)
35 > }()
36
37 > return c.client.CancelOutstandingPoll(ctx, request, opts...) metric_client_gen.go
38 }
39
224 request *matchingservice.ForceUnloadTaskQueuePartitionRequest,
225 opts ...grpc.CallOption,
226 > ) (_ *matchingservice.ForceUnloadTaskQueuePartitionResponse, retError error) { metric_client_gen.go
227 >
228 > metricsHandler, startTime := c.startMetricsRecording(ctx, "MatchingClientForceUnloadTaskQueuePartition")
229 > defer func() {
230 > c.finishMetricsRecording(metricsHandler, startTime, retError)
231 > }()
232
233 > return c.client.ForceUnloadTaskQueuePartition(ctx, request, opts...) metric_client_gen.go
234 }
235
336 request *matchingservice.RecordWorkerHeartbeatRequest,
337 opts ...grpc.CallOption,
338 > ) (_ *matchingservice.RecordWorkerHeartbeatResponse, retError error) { metric_client_gen.go
339 >
340 > metricsHandler, startTime := c.startMetricsRecording(ctx, "MatchingClientRecordWorkerHeartbeat")
341 > defer func() {
342 > c.finishMetricsRecording(metricsHandler, startTime, retError)
343 > }()
344
345 > return c.client.RecordWorkerHeartbeat(ctx, request, opts...) metric_client_gen.go
346 }
347
go.temporal.io/server/service/matching/handler.go 18 introduced LOC · 3 ranges

Open complete file

318 // CancelOutstandingPoll is used to cancel outstanding pollers
319 func (h *Handler) CancelOutstandingPoll(ctx context.Context,
320 > request *matchingservice.CancelOutstandingPollRequest) (_ *matchingservice.CancelOutstandingPollResponse, retError error) { handler.go
321 > defer log.CapturePanic(h.logger, &retError)
322 > err := h.engine.CancelOutstandingPoll(ctx, request)
323 > return &matchingservice.CancelOutstandingPollResponse{}, err
324 > }
325
326 // CancelOutstandingWorkerPolls cancels all outstanding polls for a given worker instance key.
474 ctx context.Context,
475 request *matchingservice.ForceUnloadTaskQueuePartitionRequest,
476 > ) (_ *matchingservice.ForceUnloadTaskQueuePartitionResponse, retError error) { handler.go
477 > defer log.CapturePanic(h.logger, &retError)
478 > return h.engine.ForceUnloadTaskQueuePartition(ctx, request)
479 > }
480
481 func (h *Handler) ForceLoadTaskQueuePartition(
601 func (h *Handler) RecordWorkerHeartbeat(
602 ctx context.Context, request *matchingservice.RecordWorkerHeartbeatRequest,
603 > ) (_ *matchingservice.RecordWorkerHeartbeatResponse, retError error) { handler.go
604 > defer log.CapturePanic(h.logger, &retError)
605 > nsID := namespace.ID(request.GetNamespaceId())
606 > nsName := h.namespaceName(nsID)
607 > principal := headers.GetPrincipal(ctx)
608 >
609 > h.workersRegistry.RecordWorkerHeartbeats(nsID, nsName, principal, request.GetHeartbeartRequest().GetWorkerHeartbeat())
610 > return &matchingservice.RecordWorkerHeartbeatResponse{}, nil
611 > }
612
613 // ListWorkers retrieves a list of workers in the specified namespace that match the provided filters.
go.temporal.io/server/common/rpc/dial_tracer.go 12 introduced LOC · 1 range

Open complete file

61
62 if dialErr != nil {
63 > fields := []tag.Tag{ dial_tracer.go
64 > tag.Duration("totalDuration", total),
65 > tag.Error(dialErr),
66 > tag.ErrorType(dialErr),
67 > tag.Duration("connectDuration", ndt.connectDuration),
68 > tag.String("connectAddr", ndt.connectAddr),
69 > }
70 > if ndt.connectErr != nil {
71 > fields = append(fields, tag.String("connectErr", ndt.connectErr.Error()))
72 > }
73 > d.logger.Warn("network dial error", fields...)
74 > metrics.ServiceDialErrorCount.With(d.metricsHandler).Record(1)
75 } else {
76 metrics.ServiceDialSuccessCount.With(d.metricsHandler).Record(1)
go.temporal.io/server/common/persistence/persistence_metric_clients.go 9 introduced LOC · 1 range

Open complete file

654 ctx context.Context,
655 request *UpdateTaskQueueRequest,
656 > ) (_ *UpdateTaskQueueResponse, retErr error) { persistence_metric_clients.go
657 > caller := headers.GetCallerInfo(ctx).CallerName
658 > startTime := time.Now().UTC()
659 > defer func() {
660 > p.healthSignals.Record(CallerSegmentMissing, time.Since(startTime), retErr)
661 > p.recordRequestMetrics(metrics.PersistenceUpdateTaskQueueScope, caller, time.Since(startTime), retErr)
662 > p.recordDataLossMetrics(metrics.PersistenceUpdateTaskQueueScope, caller, retErr, "", "")
663 > }()
664 > return p.persistence.UpdateTaskQueue(ctx, request)
665 }
666
go.temporal.io/server/common/persistence/persistence_retryable_clients.go 9 introduced LOC · 2 ranges

Open complete file

669 ctx context.Context,
670 request *UpdateTaskQueueRequest,
671 > ) (*UpdateTaskQueueResponse, error) { persistence_retryable_clients.go
672 > var response *UpdateTaskQueueResponse
673 > op := func(ctx context.Context) error {
674 > var err error
675 > response, err = p.persistence.UpdateTaskQueue(ctx, request)
676 > return err
677 > }
678
679 > err := backoff.ThrottleRetryContext(ctx, op, p.policy, p.isRetryable) persistence_retryable_clients.go
680 > return response, err
681 }
682
go.temporal.io/server/common/rpc/interceptor/telemetry.go 9 introduced LOC · 3 ranges

Open complete file

192
193 if err != nil {
194 > ti.requestErrorHandler.HandleError(req, info.FullMethod, metricsHandler, logTags, err, nsName) telemetry.go
195 } else {
196 // emit action metrics only after successful calls
333 }
334
335 > case pollActivityTaskQueue: telemetry.go
336 > // handle activity retries
337 > activityPollResponse, ok := result.(*workflowservice.PollActivityTaskQueueResponse)
338 > if !ok {
339 return
340 }
341 > if activityPollResponse == nil || len(activityPollResponse.TaskToken) == 0 { telemetry.go
342 > // empty response
343 > return
344 > }
345 if activityPollResponse.Attempt > 1 {
346 metrics.ActionCounter.With(metricsHandler).Record(1, metrics.ActionType("activity_retry"))
go.temporal.io/server/service/matching/matching_engine.go 8 introduced LOC · 2 ranges

Open complete file

1212 _ context.Context,
1213 request *matchingservice.CancelOutstandingPollRequest,
1214 > ) error { matching_engine.go
1215 > cancel, ok := e.outstandingPollers.Pop(request.PollerId)
1216 > if ok {
1217 > cancel()
1218 > }
1219 > return nil
1220 }
1221
2905 case <-ownershipLostCh:
2906 return nil, serviceerror.NewAborted("Nexus endpoints table ownership lost")
2907 > case <-ctx.Done(): matching_engine.go
2908 > return resp, nil
2909 case <-tableVersionChanged:
2910 continue
go.temporal.io/server/client/matching/metric_client.go 3 introduced LOC · 2 ranges

Open complete file

90 scope, stopwatch := c.startMetricsRecording(ctx, metrics.MatchingClientPollActivityTaskQueueScope)
91 defer func() {
92 > c.finishMetricsRecording(scope, stopwatch, retError) metric_client.go
93 > }()
94
95 if request.PollRequest != nil {
234 *serviceerror.NewerBuildExists,
235 *serviceerror.WorkflowExecutionAlreadyStarted,
236 > *serviceerror.ResourceExhausted: metric_client.go
237 // noop - not interest and too many logs
238 default:
go.temporal.io/server/common/persistence/persistence_rate_limited_clients.go 3 introduced LOC · 2 ranges

Open complete file

556 ctx context.Context,
557 request *UpdateTaskQueueRequest,
558 > ) (*UpdateTaskQueueResponse, error) { persistence_rate_limited_clients.go
559 > if err := allow(ctx, "UpdateTaskQueue", CallerSegmentMissing, p.systemRateLimiter, p.namespaceRateLimiter, p.shardRateLimiter); err != nil {
560 return nil, err
561 }
562 > return p.persistence.UpdateTaskQueue(ctx, request) persistence_rate_limited_clients.go
563 }
564
go.temporal.io/server/service/frontend/service.go 3 introduced LOC · 1 range

Open complete file

548 wg.Go(func() {
549 t := time.AfterFunc(requestDrainTime, func() {
550 > s.logger.Info("ShutdownHandler: Drain time expired, stopping all traffic") service.go
551 > s.server.Stop()
552 > })
553 s.server.GracefulStop()
554 t.Stop()
go.temporal.io/server/common/client_cache.go 2 introduced LOC · 1 range

Open complete file

138
139 for _, entry := range entries {
140 > if entry.release != nil { client_cache.go
141 > if err := entry.release(); err != nil {
142 c.logger.Warn("Error releasing evicted client resource", tag.Error(err))
143 }
go.temporal.io/server/common/metrics/grpc.go 2 introduced LOC · 1 range

Open complete file

90
91 select {
92 > case <-ctx.Done(): grpc.go
93 > return resp, err
94 default:
95 }
go.temporal.io/server/common/rpc/interceptor/context_metadata_interceptor.go 2 introduced LOC · 1 range

Open complete file

67 // and SetTrailer would return ErrIllegalHeaderWrite ("SendHeader called multiple times").
68 select {
69 > case <-ctx.Done(): context_metadata_interceptor.go
70 > return
71 default:
72 }
go.temporal.io/server/service/matching/simple_partition_scaler.go 2 introduced LOC · 1 range

Open complete file

116 }
117
118 > func (*simplePartitionScaler) Stop() { simple_partition_scaler.go
119 > }
120
121 func (s *simplePartitionScaler) updateAddTarget(

Direct hierarchy

Parents

Children