Vihren Atlas

handler.go ×25

Code frontier

unlabeled · c_e4c1b99b8833

10 tests · 39660 LOC · 738 files · introduces 0 tests · 728 LOC · 45 files

Introduces — evidence that enters the hierarchy at this concept

Code
210 ranges728 lines · 45 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
9209 ranges39660 lines · 738 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.

Showing the top 20 of 45 files by introduced lines: 628 of 728 introduced LOC and 165 of 210 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/api/historyservice/v1/request_response.pb.go 84 introduced LOC · 15 ranges

Open complete file

207 }
208
209 > func (x *StartWorkflowExecutionRequest) Reset() { request_response.pb.go
210 > *x = StartWorkflowExecutionRequest{}
211 > mi := &file_temporal_server_api_historyservice_v1_request_response_proto_msgTypes[1]
212 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
213 > ms.StoreMessageInfo(mi)
214 > }
215
216 func (x *StartWorkflowExecutionRequest) String() string {
380 }
381
382 > func (x *StartWorkflowExecutionResponse) Reset() { request_response.pb.go
383 > *x = StartWorkflowExecutionResponse{}
384 > mi := &file_temporal_server_api_historyservice_v1_request_response_proto_msgTypes[2]
385 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
386 > ms.StoreMessageInfo(mi)
387 > }
388
389 func (x *StartWorkflowExecutionResponse) String() string {
396 mi := &file_temporal_server_api_historyservice_v1_request_response_proto_msgTypes[2]
397 if x != nil {
398 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) request_response.pb.go
399 > if ms.LoadMessageInfo() == nil {
400 > ms.StoreMessageInfo(mi)
401 > }
402 > return ms
403 }
404 return mi.MessageOf(x)
1203 }
1204
1205 > func (x *RecordWorkflowTaskStartedRequest) Reset() { request_response.pb.go
1206 > *x = RecordWorkflowTaskStartedRequest{}
1207 > mi := &file_temporal_server_api_historyservice_v1_request_response_proto_msgTypes[11]
1208 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
1209 > ms.StoreMessageInfo(mi)
1210 > }
1211
1212 func (x *RecordWorkflowTaskStartedRequest) String() string {
1219 mi := &file_temporal_server_api_historyservice_v1_request_response_proto_msgTypes[11]
1220 if x != nil {
1221 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) request_response.pb.go
1222 > if ms.LoadMessageInfo() == nil {
1223 > ms.StoreMessageInfo(mi)
1224 > }
1225 > return ms
1226 }
1227 return mi.MessageOf(x)
1240 }
1241
1242 > func (x *RecordWorkflowTaskStartedRequest) GetWorkflowExecution() *v14.WorkflowExecution { request_response.pb.go
1243 > if x != nil {
1244 > return x.WorkflowExecution
1245 > }
1246 return nil
1247 }
1358 }
1359
1360 > func (x *RecordWorkflowTaskStartedResponse) Reset() { request_response.pb.go
1361 > *x = RecordWorkflowTaskStartedResponse{}
1362 > mi := &file_temporal_server_api_historyservice_v1_request_response_proto_msgTypes[12]
1363 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
1364 > ms.StoreMessageInfo(mi)
1365 > }
1366
1367 func (x *RecordWorkflowTaskStartedResponse) String() string {
1374 mi := &file_temporal_server_api_historyservice_v1_request_response_proto_msgTypes[12]
1375 if x != nil {
1376 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) request_response.pb.go
1377 > if ms.LoadMessageInfo() == nil {
1378 > ms.StoreMessageInfo(mi)
1379 > }
1380 > return ms
1381 }
1382 return mi.MessageOf(x)
1590 func (*RecordWorkflowTaskStartedResponseWithRawHistory) ProtoMessage() {}
1591
1592 > func (x *RecordWorkflowTaskStartedResponseWithRawHistory) ProtoReflect() protoreflect.Message { request_response.pb.go
1593 > mi := &file_temporal_server_api_historyservice_v1_request_response_proto_msgTypes[13]
1594 > if x != nil {
1595 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
1596 > if ms.LoadMessageInfo() == nil {
1597 > ms.StoreMessageInfo(mi)
1598 > }
1599 > return ms
1600 }
1601 return mi.MessageOf(x)
2039 }
2040
2041 > func (x *RespondWorkflowTaskCompletedRequest) Reset() { request_response.pb.go
2042 > *x = RespondWorkflowTaskCompletedRequest{}
2043 > mi := &file_temporal_server_api_historyservice_v1_request_response_proto_msgTypes[16]
2044 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
2045 > ms.StoreMessageInfo(mi)
2046 > }
2047
2048 func (x *RespondWorkflowTaskCompletedRequest) String() string {
2055 mi := &file_temporal_server_api_historyservice_v1_request_response_proto_msgTypes[16]
2056 if x != nil {
2057 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) request_response.pb.go
2058 > if ms.LoadMessageInfo() == nil {
2059 > ms.StoreMessageInfo(mi)
2060 > }
2061 > return ms
2062 }
2063 return mi.MessageOf(x)
2094 }
2095
2096 > func (x *RespondWorkflowTaskCompletedResponse) Reset() { request_response.pb.go
2097 > *x = RespondWorkflowTaskCompletedResponse{}
2098 > mi := &file_temporal_server_api_historyservice_v1_request_response_proto_msgTypes[17]
2099 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
2100 > ms.StoreMessageInfo(mi)
2101 > }
2102
2103 func (x *RespondWorkflowTaskCompletedResponse) String() string {
2110 mi := &file_temporal_server_api_historyservice_v1_request_response_proto_msgTypes[17]
2111 if x != nil {
2112 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) request_response.pb.go
2113 > if ms.LoadMessageInfo() == nil {
2114 > ms.StoreMessageInfo(mi)
2115 > }
2116 > return ms
2117 }
2118 return mi.MessageOf(x)
7579 }
7580
7581 > func (x *GetWorkflowExecutionHistoryRequest) Reset() { request_response.pb.go
7582 > *x = GetWorkflowExecutionHistoryRequest{}
7583 > mi := &file_temporal_server_api_historyservice_v1_request_response_proto_msgTypes[111]
7584 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
7585 > ms.StoreMessageInfo(mi)
7586 > }
7587
7588 func (x *GetWorkflowExecutionHistoryRequest) String() string {
7595 mi := &file_temporal_server_api_historyservice_v1_request_response_proto_msgTypes[111]
7596 if x != nil {
7597 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) request_response.pb.go
7598 > if ms.LoadMessageInfo() == nil {
7599 > ms.StoreMessageInfo(mi)
7600 > }
7601 > return ms
7602 }
7603 return mi.MessageOf(x)
go.temporal.io/server/api/historyservice/v1/service_grpc.pb.go 68 introduced LOC · 20 ranges

Open complete file

400 }
401
402 > func (c *historyServiceClient) StartWorkflowExecution(ctx context.Context, in *StartWorkflowExecutionRequest, opts ...grpc.CallOption) (*StartWorkflowExecutionResponse, error) { service_grpc.pb.go
403 > out := new(StartWorkflowExecutionResponse)
404 > err := c.cc.Invoke(ctx, HistoryService_StartWorkflowExecution_FullMethodName, in, out, opts...)
405 > if err != nil {
406 return nil, err
407 }
408 > return out, nil service_grpc.pb.go
409 }
410
436 }
437
438 > func (c *historyServiceClient) RecordWorkflowTaskStarted(ctx context.Context, in *RecordWorkflowTaskStartedRequest, opts ...grpc.CallOption) (*RecordWorkflowTaskStartedResponse, error) { service_grpc.pb.go
439 > out := new(RecordWorkflowTaskStartedResponse)
440 > err := c.cc.Invoke(ctx, HistoryService_RecordWorkflowTaskStarted_FullMethodName, in, out, opts...)
441 > if err != nil {
442 return nil, err
443 }
444 > return out, nil service_grpc.pb.go
445 }
446
454 }
455
456 > func (c *historyServiceClient) RespondWorkflowTaskCompleted(ctx context.Context, in *RespondWorkflowTaskCompletedRequest, opts ...grpc.CallOption) (*RespondWorkflowTaskCompletedResponse, error) { service_grpc.pb.go
457 > out := new(RespondWorkflowTaskCompletedResponse)
458 > err := c.cc.Invoke(ctx, HistoryService_RespondWorkflowTaskCompleted_FullMethodName, in, out, opts...)
459 > if err != nil {
460 return nil, err
461 }
462 > return out, nil service_grpc.pb.go
463 }
464
899 }
900
901 > func (c *historyServiceClient) GetWorkflowExecutionHistory(ctx context.Context, in *GetWorkflowExecutionHistoryRequest, opts ...grpc.CallOption) (*GetWorkflowExecutionHistoryResponse, error) { service_grpc.pb.go
902 > out := new(GetWorkflowExecutionHistoryResponse)
903 > err := c.cc.Invoke(ctx, HistoryService_GetWorkflowExecutionHistory_FullMethodName, in, out, opts...)
904 > if err != nil {
905 return nil, err
906 }
907 > return out, nil service_grpc.pb.go
908 }
909
1656 }
1657
1658 > func _HistoryService_StartWorkflowExecution_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { service_grpc.pb.go
1659 > in := new(StartWorkflowExecutionRequest)
1660 > if err := dec(in); err != nil {
1661 return nil, err
1662 }
1663 > if interceptor == nil { service_grpc.pb.go
1664 return srv.(HistoryServiceServer).StartWorkflowExecution(ctx, in)
1665 }
1666 > info := &grpc.UnaryServerInfo{ service_grpc.pb.go
1667 > Server: srv,
1668 > FullMethod: HistoryService_StartWorkflowExecution_FullMethodName,
1669 > }
1670 > handler := func(ctx context.Context, req interface{}) (interface{}, error) {
1671 > return srv.(HistoryServiceServer).StartWorkflowExecution(ctx, req.(*StartWorkflowExecutionRequest))
1672 > }
1673 > return interceptor(ctx, in, info, handler)
1674 }
1675
1728 }
1729
1730 > func _HistoryService_RecordWorkflowTaskStarted_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { service_grpc.pb.go
1731 > in := new(RecordWorkflowTaskStartedRequest)
1732 > if err := dec(in); err != nil {
1733 return nil, err
1734 }
1735 > if interceptor == nil { service_grpc.pb.go
1736 return srv.(HistoryServiceServer).RecordWorkflowTaskStarted(ctx, in)
1737 }
1738 > info := &grpc.UnaryServerInfo{ service_grpc.pb.go
1739 > Server: srv,
1740 > FullMethod: HistoryService_RecordWorkflowTaskStarted_FullMethodName,
1741 > }
1742 > handler := func(ctx context.Context, req interface{}) (interface{}, error) {
1743 > return srv.(HistoryServiceServer).RecordWorkflowTaskStarted(ctx, req.(*RecordWorkflowTaskStartedRequest))
1744 > }
1745 > return interceptor(ctx, in, info, handler)
1746 }
1747
1764 }
1765
1766 > func _HistoryService_RespondWorkflowTaskCompleted_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { service_grpc.pb.go
1767 > in := new(RespondWorkflowTaskCompletedRequest)
1768 > if err := dec(in); err != nil {
1769 return nil, err
1770 }
1771 > if interceptor == nil { service_grpc.pb.go
1772 return srv.(HistoryServiceServer).RespondWorkflowTaskCompleted(ctx, in)
1773 }
1774 > info := &grpc.UnaryServerInfo{ service_grpc.pb.go
1775 > Server: srv,
1776 > FullMethod: HistoryService_RespondWorkflowTaskCompleted_FullMethodName,
1777 > }
1778 > handler := func(ctx context.Context, req interface{}) (interface{}, error) {
1779 > return srv.(HistoryServiceServer).RespondWorkflowTaskCompleted(ctx, req.(*RespondWorkflowTaskCompletedRequest))
1780 > }
1781 > return interceptor(ctx, in, info, handler)
1782 }
1783
2618 }
2619
2620 > func _HistoryService_GetWorkflowExecutionHistory_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { service_grpc.pb.go
2621 > in := new(GetWorkflowExecutionHistoryRequest)
2622 > if err := dec(in); err != nil {
2623 return nil, err
2624 }
2625 > if interceptor == nil { service_grpc.pb.go
2626 return srv.(HistoryServiceServer).GetWorkflowExecutionHistory(ctx, in)
2627 }
2628 > info := &grpc.UnaryServerInfo{ service_grpc.pb.go
2629 > Server: srv,
2630 > FullMethod: HistoryService_GetWorkflowExecutionHistory_FullMethodName,
2631 > }
2632 > handler := func(ctx context.Context, req interface{}) (interface{}, error) {
2633 > return srv.(HistoryServiceServer).GetWorkflowExecutionHistory(ctx, req.(*GetWorkflowExecutionHistoryRequest))
2634 > }
2635 > return interceptor(ctx, in, info, handler)
2636 }
2637
go.temporal.io/server/service/history/handler.go 65 introduced LOC · 25 ranges

Open complete file

356
357 // RecordWorkflowTaskStarted - Record Workflow Task started.
358 > func (h *Handler) RecordWorkflowTaskStarted(ctx context.Context, request *historyservice.RecordWorkflowTaskStartedRequest) (*historyservice.RecordWorkflowTaskStartedResponseWithRawHistory, error) { handler.go
359 > namespaceID := namespace.ID(request.GetNamespaceId())
360 > workflowExecution := request.WorkflowExecution
361 > workflowID := workflowExecution.GetWorkflowId()
362 > if namespaceID == "" {
363 return nil, h.convertError(errNamespaceNotSet)
364 }
365
366 > if request.PollRequest == nil || request.PollRequest.TaskQueue.GetName() == "" { handler.go
367 return nil, h.convertError(errTaskQueueNotSet)
368 }
369
370 > shardContext, err := h.controller.GetShardByNamespaceWorkflow(namespaceID, workflowID) handler.go
371 > if err != nil {
372 return nil, h.convertError(err)
373 }
374 > engine, err := shardContext.GetEngine(ctx) handler.go
375 > if err != nil {
376 h.logger.Error("RecordWorkflowTaskStarted failed.",
377 tag.Error(err),
383 }
384
385 > response, err := engine.RecordWorkflowTaskStarted(ctx, request) handler.go
386 > if err != nil {
387 return nil, h.convertError(err)
388 }
389 > response.Clock, err = shardContext.NewVectorClock() handler.go
390 > if err != nil {
391 return nil, h.convertError(err)
392 }
393 > return response, nil handler.go
394 }
395
548
549 // RespondWorkflowTaskCompleted - records completion of a workflow task
550 > func (h *Handler) RespondWorkflowTaskCompleted(ctx context.Context, request *historyservice.RespondWorkflowTaskCompletedRequest) (*historyservice.RespondWorkflowTaskCompletedResponse, error) { handler.go
551 > namespaceID := namespace.ID(request.GetNamespaceId())
552 > if namespaceID == "" {
553 return nil, h.convertError(errNamespaceNotSet)
554 }
555
556 > completeRequest := request.CompleteRequest handler.go
557 > token, err := h.tokenSerializer.Deserialize(completeRequest.TaskToken)
558 > if err != nil {
559 return nil, consts.ErrDeserializingToken
560 }
561
562 > h.logger.Debug("RespondWorkflowTaskCompleted", handler.go
563 > tag.WorkflowNamespaceID(token.GetNamespaceId()),
564 > tag.WorkflowID(token.GetWorkflowId()),
565 > tag.WorkflowRunID(token.GetRunId()),
566 > tag.WorkflowScheduledEventID(token.GetScheduledEventId()))
567 >
568 > err = validateTaskToken(token)
569 > if err != nil {
570 return nil, h.convertError(err)
571 }
572 > workflowID := token.GetWorkflowId() handler.go
573 >
574 > shardContext, err := h.controller.GetShardByNamespaceWorkflow(namespaceID, workflowID)
575 > if err != nil {
576 return nil, h.convertError(err)
577 }
578 > engine, err := shardContext.GetEngine(ctx) handler.go
579 > if err != nil {
580 return nil, h.convertError(err)
581 }
582
583 > response, err := engine.RespondWorkflowTaskCompleted(ctx, request) handler.go
584 > if err != nil {
585 return nil, h.convertError(err)
586 }
587
588 > return response, nil handler.go
589 }
590
632
633 // StartWorkflowExecution - creates a new workflow execution
634 > func (h *Handler) StartWorkflowExecution(ctx context.Context, request *historyservice.StartWorkflowExecutionRequest) (*historyservice.StartWorkflowExecutionResponse, error) { handler.go
635 > namespaceID := namespace.ID(request.GetNamespaceId())
636 > if namespaceID == "" {
637 return nil, h.convertError(errNamespaceNotSet)
638 }
639
640 > startRequest := request.StartRequest handler.go
641 > workflowID := startRequest.GetWorkflowId()
642 > shardContext, err := h.controller.GetShardByNamespaceWorkflow(namespaceID, workflowID)
643 > if err != nil {
644 return nil, h.convertError(err)
645 }
646
647 > engine, err := shardContext.GetEngine(ctx) handler.go
648 > if err != nil {
649 return nil, h.convertError(err)
650 }
651
652 > response, err := engine.StartWorkflowExecution(ctx, request) handler.go
653 > if err != nil {
654 return nil, h.convertError(err)
655 }
656 > if response.Clock == nil { handler.go
657 > response.Clock, err = shardContext.NewVectorClock()
658 > if err != nil {
659 return nil, h.convertError(err)
660 }
661 }
662 > return response, nil handler.go
663 }
664
1926 ctx context.Context,
1927 request *historyservice.GetWorkflowExecutionHistoryRequest,
1928 > ) (*historyservice.GetWorkflowExecutionHistoryResponseWithRaw, error) { handler.go
1929 > shardContext, err := h.controller.GetShardByNamespaceWorkflow(
1930 > namespace.ID(request.GetNamespaceId()),
1931 > request.Request.GetExecution().GetWorkflowId(),
1932 > )
1933 > if err != nil {
1934 return nil, h.convertError(err)
1935 }
1936
1937 > engine, err := shardContext.GetEngine(ctx) handler.go
1938 > if err != nil {
1939 return nil, h.convertError(err)
1940 }
1941
1942 > return engine.GetWorkflowExecutionHistory(ctx, request) handler.go
1943 }
1944
2260 }
2261
2262 > func validateTaskToken(taskToken *tokenspb.Task) error { handler.go
2263 > if len(taskToken.GetComponentRef()) == 0 && taskToken.GetWorkflowId() == "" {
2264 return errBusinessIDNotSet
2265 }
2266
2267 > return nil handler.go
2268 }
2269
go.temporal.io/server/client/history/client_gen.go 58 introduced LOC · 10 ranges

Open complete file

404 request *historyservice.GetWorkflowExecutionHistoryRequest,
405 opts ...grpc.CallOption,
406 > ) (*historyservice.GetWorkflowExecutionHistoryResponse, error) { client_gen.go
407 > shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetRequest().GetExecution().GetWorkflowId())
408 > var response *historyservice.GetWorkflowExecutionHistoryResponse
409 > op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
410 > var err error
411 > ctx, cancel := c.createContext(ctx)
412 > defer cancel()
413 > response, err = client.GetWorkflowExecutionHistory(ctx, request, opts...)
414 > return err
415 > }
416 > if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
417 return nil, err
418 }
419 > return response, nil client_gen.go
420 }
421
842 request *historyservice.RecordWorkflowTaskStartedRequest,
843 opts ...grpc.CallOption,
844 > ) (*historyservice.RecordWorkflowTaskStartedResponse, error) { client_gen.go
845 > shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetWorkflowExecution().GetWorkflowId())
846 > var response *historyservice.RecordWorkflowTaskStartedResponse
847 > op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
848 > var err error
849 > ctx, cancel := c.createContext(ctx)
850 > defer cancel()
851 > response, err = client.RecordWorkflowTaskStarted(ctx, request, opts...)
852 > return err
853 > }
854 > if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
855 return nil, err
856 }
857 > return response, nil client_gen.go
858 }
859
1156 request *historyservice.RespondWorkflowTaskCompletedRequest,
1157 opts ...grpc.CallOption,
1158 > ) (*historyservice.RespondWorkflowTaskCompletedResponse, error) { client_gen.go
1159 > taskToken, err := c.tokenSerializer.Deserialize(request.GetCompleteRequest().GetTaskToken())
1160 > if err != nil {
1161 return nil, serviceerror.NewInvalidArgument("error deserializing task token")
1162 }
1163 > var namespaceID string client_gen.go
1164 > var businessID string
1165 > if len(taskToken.GetComponentRef()) > 0 {
1166 ref, err := c.tokenSerializer.DeserializeChasmComponentRef(taskToken.GetComponentRef())
1167 if err != nil {
1170 namespaceID = ref.GetNamespaceId()
1171 businessID = ref.GetBusinessId()
1172 > } else { client_gen.go
1173 > namespaceID = request.GetNamespaceId()
1174 > businessID = taskToken.GetWorkflowId()
1175 > }
1176 > shardID := c.shardIDFromWorkflowID(namespaceID, businessID)
1177 >
1178 > var response *historyservice.RespondWorkflowTaskCompletedResponse
1179 > op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
1180 > var err error
1181 > ctx, cancel := c.createContext(ctx)
1182 > defer cancel()
1183 > response, err = client.RespondWorkflowTaskCompleted(ctx, request, opts...)
1184 > return err
1185 > }
1186 > if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
1187 return nil, err
1188 }
1189 > return response, nil client_gen.go
1190 }
1191
1312 request *historyservice.StartWorkflowExecutionRequest,
1313 opts ...grpc.CallOption,
1314 > ) (*historyservice.StartWorkflowExecutionResponse, error) { client_gen.go
1315 > shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetStartRequest().GetWorkflowId())
1316 > var response *historyservice.StartWorkflowExecutionResponse
1317 > op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
1318 > var err error
1319 > ctx, cancel := c.createContext(ctx)
1320 > defer cancel()
1321 > response, err = client.StartWorkflowExecution(ctx, request, opts...)
1322 > return err
1323 > }
1324 > if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
1325 return nil, err
1326 }
1327 > return response, nil client_gen.go
1328 }
1329
go.temporal.io/server/chasm/lib/nexusoperation/gen/nexusoperationpb/v1/request_response.pb.go 48 introduced LOC · 24 ranges

Open complete file

46 func (*StartNexusOperationRequest) ProtoMessage() {}
47
48 > func (x *StartNexusOperationRequest) ProtoReflect() protoreflect.Message { request_response.pb.go
49 > mi := &file_temporal_server_chasm_lib_nexusoperation_proto_v1_request_response_proto_msgTypes[0]
50 > if x != nil {
51 ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
52 if ms.LoadMessageInfo() == nil {
104 func (*StartNexusOperationResponse) ProtoMessage() {}
105
106 > func (x *StartNexusOperationResponse) ProtoReflect() protoreflect.Message { request_response.pb.go
107 > mi := &file_temporal_server_chasm_lib_nexusoperation_proto_v1_request_response_proto_msgTypes[1]
108 > if x != nil {
109 ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
110 if ms.LoadMessageInfo() == nil {
149 func (*DescribeNexusOperationRequest) ProtoMessage() {}
150
151 > func (x *DescribeNexusOperationRequest) ProtoReflect() protoreflect.Message { request_response.pb.go
152 > mi := &file_temporal_server_chasm_lib_nexusoperation_proto_v1_request_response_proto_msgTypes[2]
153 > if x != nil {
154 ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
155 if ms.LoadMessageInfo() == nil {
200 func (*DescribeNexusOperationResponse) ProtoMessage() {}
201
202 > func (x *DescribeNexusOperationResponse) ProtoReflect() protoreflect.Message { request_response.pb.go
203 > mi := &file_temporal_server_chasm_lib_nexusoperation_proto_v1_request_response_proto_msgTypes[3]
204 > if x != nil {
205 ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
206 if ms.LoadMessageInfo() == nil {
245 func (*RequestCancelNexusOperationRequest) ProtoMessage() {}
246
247 > func (x *RequestCancelNexusOperationRequest) ProtoReflect() protoreflect.Message { request_response.pb.go
248 > mi := &file_temporal_server_chasm_lib_nexusoperation_proto_v1_request_response_proto_msgTypes[4]
249 > if x != nil {
250 ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
251 if ms.LoadMessageInfo() == nil {
295 func (*RequestCancelNexusOperationResponse) ProtoMessage() {}
296
297 > func (x *RequestCancelNexusOperationResponse) ProtoReflect() protoreflect.Message { request_response.pb.go
298 > mi := &file_temporal_server_chasm_lib_nexusoperation_proto_v1_request_response_proto_msgTypes[5]
299 > if x != nil {
300 ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
301 if ms.LoadMessageInfo() == nil {
333 func (*TerminateNexusOperationRequest) ProtoMessage() {}
334
335 > func (x *TerminateNexusOperationRequest) ProtoReflect() protoreflect.Message { request_response.pb.go
336 > mi := &file_temporal_server_chasm_lib_nexusoperation_proto_v1_request_response_proto_msgTypes[6]
337 > if x != nil {
338 ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
339 if ms.LoadMessageInfo() == nil {
383 func (*TerminateNexusOperationResponse) ProtoMessage() {}
384
385 > func (x *TerminateNexusOperationResponse) ProtoReflect() protoreflect.Message { request_response.pb.go
386 > mi := &file_temporal_server_chasm_lib_nexusoperation_proto_v1_request_response_proto_msgTypes[7]
387 > if x != nil {
388 ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
389 if ms.LoadMessageInfo() == nil {
421 func (*DeleteNexusOperationRequest) ProtoMessage() {}
422
423 > func (x *DeleteNexusOperationRequest) ProtoReflect() protoreflect.Message { request_response.pb.go
424 > mi := &file_temporal_server_chasm_lib_nexusoperation_proto_v1_request_response_proto_msgTypes[8]
425 > if x != nil {
426 ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
427 if ms.LoadMessageInfo() == nil {
471 func (*DeleteNexusOperationResponse) ProtoMessage() {}
472
473 > func (x *DeleteNexusOperationResponse) ProtoReflect() protoreflect.Message { request_response.pb.go
474 > mi := &file_temporal_server_chasm_lib_nexusoperation_proto_v1_request_response_proto_msgTypes[9]
475 > if x != nil {
476 ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
477 if ms.LoadMessageInfo() == nil {
509 func (*PollNexusOperationRequest) ProtoMessage() {}
510
511 > func (x *PollNexusOperationRequest) ProtoReflect() protoreflect.Message { request_response.pb.go
512 > mi := &file_temporal_server_chasm_lib_nexusoperation_proto_v1_request_response_proto_msgTypes[10]
513 > if x != nil {
514 ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
515 if ms.LoadMessageInfo() == nil {
560 func (*PollNexusOperationResponse) ProtoMessage() {}
561
562 > func (x *PollNexusOperationResponse) ProtoReflect() protoreflect.Message { request_response.pb.go
563 > mi := &file_temporal_server_chasm_lib_nexusoperation_proto_v1_request_response_proto_msgTypes[11]
564 > if x != nil {
565 ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
566 if ms.LoadMessageInfo() == nil {
go.temporal.io/server/client/history/retryable_client_gen.go 36 introduced LOC · 4 ranges

Open complete file

361 request *historyservice.GetWorkflowExecutionHistoryRequest,
362 opts ...grpc.CallOption,
363 > ) (*historyservice.GetWorkflowExecutionHistoryResponse, error) { retryable_client_gen.go
364 > var resp *historyservice.GetWorkflowExecutionHistoryResponse
365 > op := func(ctx context.Context) error {
366 > var err error
367 > resp, err = c.client.GetWorkflowExecutionHistory(ctx, request, opts...)
368 > return err
369 > }
370 > err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
371 > return resp, err
372 }
373
691 request *historyservice.RecordWorkflowTaskStartedRequest,
692 opts ...grpc.CallOption,
693 > ) (*historyservice.RecordWorkflowTaskStartedResponse, error) { retryable_client_gen.go
694 > var resp *historyservice.RecordWorkflowTaskStartedResponse
695 > op := func(ctx context.Context) error {
696 > var err error
697 > resp, err = c.client.RecordWorkflowTaskStarted(ctx, request, opts...)
698 > return err
699 > }
700 > err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
701 > return resp, err
702 }
703
886 request *historyservice.RespondWorkflowTaskCompletedRequest,
887 opts ...grpc.CallOption,
888 > ) (*historyservice.RespondWorkflowTaskCompletedResponse, error) { retryable_client_gen.go
889 > var resp *historyservice.RespondWorkflowTaskCompletedResponse
890 > op := func(ctx context.Context) error {
891 > var err error
892 > resp, err = c.client.RespondWorkflowTaskCompleted(ctx, request, opts...)
893 > return err
894 > }
895 > err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
896 > return resp, err
897 }
898
976 request *historyservice.StartWorkflowExecutionRequest,
977 opts ...grpc.CallOption,
978 > ) (*historyservice.StartWorkflowExecutionResponse, error) { retryable_client_gen.go
979 > var resp *historyservice.StartWorkflowExecutionResponse
980 > op := func(ctx context.Context) error {
981 > var err error
982 > resp, err = c.client.StartWorkflowExecution(ctx, request, opts...)
983 > return err
984 > }
985 > err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
986 > return resp, err
987 }
988
go.temporal.io/server/common/persistence/persistence_metric_clients.go 32 introduced LOC · 6 ranges

Open complete file

229 ctx context.Context,
230 request *CreateWorkflowExecutionRequest,
231 > ) (_ *CreateWorkflowExecutionResponse, retErr error) { persistence_metric_clients.go
232 > caller := headers.GetCallerInfo(ctx).CallerName
233 > startTime := time.Now().UTC()
234 > defer func() {
235 > p.healthSignals.Record(request.ShardID, time.Since(startTime), retErr)
236 > var workflowID, runID string
237 > if request != nil {
238 > if request.NewWorkflowSnapshot.ExecutionInfo != nil {
239 > workflowID = request.NewWorkflowSnapshot.ExecutionInfo.WorkflowId
240 > }
241 > if request.NewWorkflowSnapshot.ExecutionState != nil {
242 > runID = request.NewWorkflowSnapshot.ExecutionState.RunId
243 > }
244 }
245 > p.recordRequestMetrics(metrics.PersistenceCreateWorkflowExecutionScope, caller, time.Since(startTime), retErr) persistence_metric_clients.go
246 > p.recordDataLossMetrics(metrics.PersistenceCreateWorkflowExecutionScope, caller, retErr, workflowID, runID)
247 }()
248 > return p.persistence.CreateWorkflowExecution(ctx, request) persistence_metric_clients.go
249 }
250
294 ctx context.Context,
295 request *UpdateWorkflowExecutionRequest,
296 > ) (_ *UpdateWorkflowExecutionResponse, retErr error) { persistence_metric_clients.go
297 > caller := headers.GetCallerInfo(ctx).CallerName
298 > startTime := time.Now().UTC()
299 > defer func() {
300 > p.healthSignals.Record(request.ShardID, time.Since(startTime), retErr)
301 > var workflowID, runID string
302 > if request != nil {
303 > if request.UpdateWorkflowMutation.ExecutionInfo != nil {
304 > workflowID = request.UpdateWorkflowMutation.ExecutionInfo.WorkflowId
305 > }
306 > if request.UpdateWorkflowMutation.ExecutionState != nil {
307 > runID = request.UpdateWorkflowMutation.ExecutionState.RunId
308 > }
309 }
310 > p.recordRequestMetrics(metrics.PersistenceUpdateWorkflowExecutionScope, caller, time.Since(startTime), retErr) persistence_metric_clients.go
311 > p.recordDataLossMetrics(metrics.PersistenceUpdateWorkflowExecutionScope, caller, retErr, workflowID, runID)
312 }()
313 > return p.persistence.UpdateWorkflowExecution(ctx, request) persistence_metric_clients.go
314 }
315
go.temporal.io/server/service/frontend/workflow_handler.go 29 introduced LOC · 7 ranges

Open complete file

954
955 if request.GetMaximumPageSize() <= 0 {
956 > request.MaximumPageSize = int32(wh.config.HistoryMaxPageSize(request.GetNamespace())) workflow_handler.go
957 > }
958
959 enums.SetDefaultHistoryEventFilterType(&request.HistoryEventFilterType)
1201 ctx context.Context,
1202 request *workflowservice.RespondWorkflowTaskCompletedRequest,
1203 > ) (_ *workflowservice.RespondWorkflowTaskCompletedResponse, retError error) { workflow_handler.go
1204 > defer log.CapturePanic(wh.logger, &retError)
1205 >
1206 > if request == nil {
1207 return nil, errRequestNotSet
1208 }
1209
1210 > if len(request.GetIdentity()) > wh.config.MaxIDLengthLimit() { workflow_handler.go
1211 return nil, errIdentityTooLong
1212 }
1213
1214 //nolint:staticcheck // SA1019: worker versioning v0.31
1215 > if err := wh.validateVersioningInfo( workflow_handler.go
1216 > request.Namespace,
1217 > request.WorkerVersionStamp,
1218 > request.DeploymentOptions,
1219 > request.StickyAttributes.GetWorkerTaskQueue(),
1220 > ); err != nil {
1221 return nil, err
1222 }
1223
1224 > wh.overrides.DisableEagerActivityDispatchForBuggyClients(ctx, request) workflow_handler.go
1225 >
1226 > namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace()))
1227 > if err != nil {
1228 return nil, err
1229 }
1230
1231 > response, err := wh.historyClient.RespondWorkflowTaskCompleted(ctx, workflow_handler.go
1232 > &historyservice.RespondWorkflowTaskCompletedRequest{
1233 > NamespaceId: namespaceID.String(),
1234 > CompleteRequest: request,
1235 > },
1236 > )
1237 > if err != nil {
1238 return nil, err
1239 }
1240
1241 > return &workflowservice.RespondWorkflowTaskCompletedResponse{ workflow_handler.go
1242 > WorkflowTask: response.NewWorkflowTask,
1243 > ActivityTasks: response.ActivityTasks,
1244 > ResetHistoryEventId: response.ResetHistoryEventId,
1245 > }, nil
1246 }
1247
go.temporal.io/server/client/history/metric_client_gen.go 28 introduced LOC · 8 ranges

Open complete file

322 request *historyservice.GetWorkflowExecutionHistoryRequest,
323 opts ...grpc.CallOption,
324 > ) (_ *historyservice.GetWorkflowExecutionHistoryResponse, retError error) { metric_client_gen.go
325 >
326 > metricsHandler, startTime := c.startMetricsRecording(ctx, "HistoryClientGetWorkflowExecutionHistory")
327 > defer func() {
328 > c.finishMetricsRecording(metricsHandler, startTime, retError)
329 > }()
330
331 > return c.client.GetWorkflowExecutionHistory(ctx, request, opts...) metric_client_gen.go
332 }
333
630 request *historyservice.RecordWorkflowTaskStartedRequest,
631 opts ...grpc.CallOption,
632 > ) (_ *historyservice.RecordWorkflowTaskStartedResponse, retError error) { metric_client_gen.go
633 >
634 > metricsHandler, startTime := c.startMetricsRecording(ctx, "HistoryClientRecordWorkflowTaskStarted")
635 > defer func() {
636 > c.finishMetricsRecording(metricsHandler, startTime, retError)
637 > }()
638
639 > return c.client.RecordWorkflowTaskStarted(ctx, request, opts...) metric_client_gen.go
640 }
641
812 request *historyservice.RespondWorkflowTaskCompletedRequest,
813 opts ...grpc.CallOption,
814 > ) (_ *historyservice.RespondWorkflowTaskCompletedResponse, retError error) { metric_client_gen.go
815 >
816 > metricsHandler, startTime := c.startMetricsRecording(ctx, "HistoryClientRespondWorkflowTaskCompleted")
817 > defer func() {
818 > c.finishMetricsRecording(metricsHandler, startTime, retError)
819 > }()
820
821 > return c.client.RespondWorkflowTaskCompleted(ctx, request, opts...) metric_client_gen.go
822 }
823
882 request *historyservice.StartWorkflowExecutionRequest,
883 opts ...grpc.CallOption,
884 > ) (_ *historyservice.StartWorkflowExecutionResponse, retError error) { metric_client_gen.go
885 >
886 > metricsHandler, startTime := c.startMetricsRecording(ctx, "HistoryClientStartWorkflowExecution")
887 > defer func() {
888 > c.finishMetricsRecording(metricsHandler, startTime, retError)
889 > }()
890
891 > return c.client.StartWorkflowExecution(ctx, request, opts...) metric_client_gen.go
892 }
893
go.temporal.io/server/client/matching/client.go 28 introduced LOC · 5 ranges

Open complete file

195 ctx context.Context,
196 request *matchingservice.AddWorkflowTaskRequest,
197 > opts ...grpc.CallOption) (*matchingservice.AddWorkflowTaskResponse, error) { client.go
198 > if !isPartitionAwareKind(request.GetTaskQueue().GetKind()) {
199 return c.addWorkflowTask(ctx, PartitionCounts{}, request, opts)
200 }
201 > pkey := c.partitionCache.makeKey( client.go
202 > request.GetNamespaceId(),
203 > request.GetTaskQueue().GetName(),
204 > enumspb.TASK_QUEUE_TYPE_WORKFLOW,
205 > )
206 > return invokeWithPartitionCounts(ctx, c.logger, c.partitionCache, pkey, request, opts, c.addWorkflowTask)
207 }
208
212 request *matchingservice.AddWorkflowTaskRequest,
213 opts []grpc.CallOption,
214 > ) (*matchingservice.AddWorkflowTaskResponse, error) { client.go
215 > request = common.CloneProto(request)
216 > client, err := c.pickClientForWrite(
217 > request.GetTaskQueue(),
218 > request.GetNamespaceId(),
219 > enumspb.TASK_QUEUE_TYPE_WORKFLOW,
220 > request.GetForwardInfo().GetSourcePartition(),
221 > pc,
222 > )
223 > if err != nil {
224 return nil, err
225 }
226 > ctx, cancel := c.createContext(ctx) client.go
227 > defer cancel()
228 > return client.AddWorkflowTask(ctx, request, opts...)
229 }
230
473 forwardedFrom string,
474 pc PartitionCounts,
475 > ) (matchingservice.MatchingServiceClient, error) { client.go
476 > p, tq := c.processInputPartition(proto, nsid, taskType, forwardedFrom)
477 > if tq != nil {
478 > p = c.loadBalancer.PickWritePartition(tq, pc)
479 > }
480 > proto.Name = p.RpcName()
481 > return c.getClientForTaskQueuePartition(p)
482 }
483
go.temporal.io/server/api/matchingservice/v1/request_response.pb.go 20 introduced LOC · 3 ranges

Open complete file

859 }
860
861 > func (x *AddWorkflowTaskRequest) Reset() { request_response.pb.go
862 > *x = AddWorkflowTaskRequest{}
863 > mi := &file_temporal_server_api_matchingservice_v1_request_response_proto_msgTypes[5]
864 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
865 > ms.StoreMessageInfo(mi)
866 > }
867
868 func (x *AddWorkflowTaskRequest) String() string {
968 }
969
970 > func (x *AddWorkflowTaskResponse) Reset() { request_response.pb.go
971 > *x = AddWorkflowTaskResponse{}
972 > mi := &file_temporal_server_api_matchingservice_v1_request_response_proto_msgTypes[6]
973 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
974 > ms.StoreMessageInfo(mi)
975 > }
976
977 func (x *AddWorkflowTaskResponse) String() string {
981 func (*AddWorkflowTaskResponse) ProtoMessage() {}
982
983 > func (x *AddWorkflowTaskResponse) ProtoReflect() protoreflect.Message { request_response.pb.go
984 > mi := &file_temporal_server_api_matchingservice_v1_request_response_proto_msgTypes[6]
985 > if x != nil {
986 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
987 > if ms.LoadMessageInfo() == nil {
988 > ms.StoreMessageInfo(mi)
989 > }
990 > return ms
991 }
992 return mi.MessageOf(x)
go.temporal.io/server/common/persistence/persistence_rate_limited_clients.go 19 introduced LOC · 9 ranges

Open complete file

270 }
271
272 > func (p *executionRateLimitedPersistenceClient) GetHistoryBranchUtil() HistoryBranchUtil { persistence_rate_limited_clients.go
273 > return p.persistence.GetHistoryBranchUtil()
274 > }
275
276 func (p *executionRateLimitedPersistenceClient) CreateWorkflowExecution(
277 ctx context.Context,
278 request *CreateWorkflowExecutionRequest,
279 > ) (*CreateWorkflowExecutionResponse, error) { persistence_rate_limited_clients.go
280 > if err := allow(ctx, "CreateWorkflowExecution", request.ShardID, p.systemRateLimiter, p.namespaceRateLimiter, p.shardRateLimiter); err != nil {
281 return nil, err
282 }
283
284 > response, err := p.persistence.CreateWorkflowExecution(ctx, request) persistence_rate_limited_clients.go
285 > return response, err
286 }
287
289 ctx context.Context,
290 request *GetWorkflowExecutionRequest,
291 > ) (*GetWorkflowExecutionResponse, error) { persistence_rate_limited_clients.go
292 > if err := allow(ctx, "GetWorkflowExecution", request.ShardID, p.systemRateLimiter, p.namespaceRateLimiter, p.shardRateLimiter); err != nil {
293 return nil, err
294 }
295
296 > response, err := p.persistence.GetWorkflowExecution(ctx, request) persistence_rate_limited_clients.go
297 > return response, err
298 }
299
313 ctx context.Context,
314 request *UpdateWorkflowExecutionRequest,
315 > ) (*UpdateWorkflowExecutionResponse, error) { persistence_rate_limited_clients.go
316 > if err := allow(ctx, "UpdateWorkflowExecution", request.ShardID, p.systemRateLimiter, p.namespaceRateLimiter, p.shardRateLimiter); err != nil {
317 return nil, err
318 }
319
320 > resp, err := p.persistence.UpdateWorkflowExecution(ctx, request) persistence_rate_limited_clients.go
321 > return resp, err
322 }
323
785 ctx context.Context,
786 request *ReadHistoryBranchRequest,
787 > ) (*ReadHistoryBranchResponse, error) { persistence_rate_limited_clients.go
788 > if err := allow(ctx, "ReadHistoryBranch", request.ShardID, p.systemRateLimiter, p.namespaceRateLimiter, p.shardRateLimiter); err != nil {
789 return nil, err
790 }
791 > response, err := p.persistence.ReadHistoryBranch(ctx, request) persistence_rate_limited_clients.go
792 > return response, err
793 }
794
go.temporal.io/server/common/rpc/interceptor/health_check.go 19 introduced LOC · 6 ranges

Open complete file

128 info *grpc.UnaryServerInfo,
129 handler grpc.UnaryHandler,
130 > ) (any, error) { health_check.go
131 > startTime := time.Now()
132 > resp, err := handler(ctx, req)
133 > elapsed := time.Since(startTime)
134 >
135 > // Skip health signal recording for non-standard APIs
136 > if isExcludedAPI(info.FullMethod) {
137 return resp, err
138 }
139
140 > if specialCaseAPIIsPolling(req) { health_check.go
141 > return resp, err
142 > }
143
144 // Record health signal for standard APIs
145 > h.healthSignalAggregator.Record(elapsed, err) health_check.go
146 > return resp, err
147 }
148
150 // Note that this interceptor may run in multiple Temporal services, so it needs to handle every version of
151 // each special request type. (for example, historyservice GetWorkflowExecutionHistory vs. workflowservice)
152 > func specialCaseAPIIsPolling(req any) bool { health_check.go
153 > switch request := req.(type) {
154 // history
155 > case *historyservice.GetWorkflowExecutionHistoryRequest: health_check.go
156 > inner := request.GetRequest()
157 > return inner != nil && inner.GetWaitNewEvent()
158
159 // frontend
162 case *workflowservice.DescribeActivityExecutionRequest:
163 return len(request.GetLongPollToken()) > 0
164 > default: health_check.go
165 > return false
166 }
167 }
go.temporal.io/server/common/persistence/persistence_retryable_clients.go 18 introduced LOC · 4 ranges

Open complete file

223 ctx context.Context,
224 request *GetWorkflowExecutionRequest,
225 > ) (*GetWorkflowExecutionResponse, error) { persistence_retryable_clients.go
226 > var response *GetWorkflowExecutionResponse
227 > op := func(ctx context.Context) error {
228 > var err error
229 > response, err = p.persistence.GetWorkflowExecution(ctx, request)
230 > return err
231 > }
232
233 > err := backoff.ThrottleRetryContext(ctx, op, p.policy, p.isRetryable) persistence_retryable_clients.go
234 > return response, err
235 }
236
253 ctx context.Context,
254 request *UpdateWorkflowExecutionRequest,
255 > ) (*UpdateWorkflowExecutionResponse, error) { persistence_retryable_clients.go
256 > var response *UpdateWorkflowExecutionResponse
257 > op := func(ctx context.Context) error {
258 > var err error
259 > response, err = p.persistence.UpdateWorkflowExecution(ctx, request)
260 > return err
261 > }
262
263 > err := backoff.ThrottleRetryContext(ctx, op, p.policy, p.isRetryable) persistence_retryable_clients.go
264 > return response, err
265 }
266
go.temporal.io/server/api/matchingservice/v1/service_grpc.pb.go 17 introduced LOC · 5 ranges

Open complete file

281 }
282
283 > func (c *matchingServiceClient) AddWorkflowTask(ctx context.Context, in *AddWorkflowTaskRequest, opts ...grpc.CallOption) (*AddWorkflowTaskResponse, error) { service_grpc.pb.go
284 > out := new(AddWorkflowTaskResponse)
285 > err := c.cc.Invoke(ctx, MatchingService_AddWorkflowTask_FullMethodName, in, out, opts...)
286 > if err != nil {
287 return nil, err
288 }
289 > return out, nil service_grpc.pb.go
290 }
291
1011 }
1012
1013 > func _MatchingService_AddWorkflowTask_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { service_grpc.pb.go
1014 > in := new(AddWorkflowTaskRequest)
1015 > if err := dec(in); err != nil {
1016 return nil, err
1017 }
1018 > if interceptor == nil { service_grpc.pb.go
1019 return srv.(MatchingServiceServer).AddWorkflowTask(ctx, in)
1020 }
1021 > info := &grpc.UnaryServerInfo{ service_grpc.pb.go
1022 > Server: srv,
1023 > FullMethod: MatchingService_AddWorkflowTask_FullMethodName,
1024 > }
1025 > handler := func(ctx context.Context, req interface{}) (interface{}, error) {
1026 > return srv.(MatchingServiceServer).AddWorkflowTask(ctx, req.(*AddWorkflowTaskRequest))
1027 > }
1028 > return interceptor(ctx, in, info, handler)
1029 }
1030
go.temporal.io/server/service/matching/handler.go 14 introduced LOC · 3 ranges

Open complete file

202 ctx context.Context,
203 request *matchingservice.AddWorkflowTaskRequest,
204 > ) (_ *matchingservice.AddWorkflowTaskResponse, retError error) { handler.go
205 > defer log.CapturePanic(h.logger, &retError)
206 > startT := time.Now().UTC()
207 > opMetrics := h.opMetricsHandler(
208 > request.GetNamespaceId(),
209 > request.GetTaskQueue(),
210 > enumspb.TASK_QUEUE_TYPE_WORKFLOW,
211 > metrics.MatchingAddWorkflowTaskScope,
212 > )
213 >
214 > if request.GetForwardInfo() != nil {
215 h.reportForwardedPerTaskQueueCounter(opMetrics, namespace.ID(request.GetNamespaceId()))
216 }
217
218 > assignedBuildId, syncMatch, err := h.engine.AddWorkflowTask(ctx, request) handler.go
219 > if syncMatch {
220 metrics.SyncMatchLatencyPerTaskQueue.With(opMetrics).Record(time.Since(startT))
221 }
222 > return &matchingservice.AddWorkflowTaskResponse{AssignedBuildId: assignedBuildId}, err handler.go
223 }
224
go.temporal.io/server/client/matching/metric_client.go 13 introduced LOC · 2 ranges

Open complete file

66 request *matchingservice.AddWorkflowTaskRequest,
67 opts ...grpc.CallOption,
68 > ) (_ *matchingservice.AddWorkflowTaskResponse, retError error) { metric_client.go
69 >
70 > scope, stopwatch := c.startMetricsRecording(ctx, metrics.MatchingClientAddWorkflowTaskScope)
71 > defer func() {
72 > c.finishMetricsRecording(scope, stopwatch, retError)
73 > }()
74
75 > c.emitForwardedSourceStats( metric_client.go
76 > scope,
77 > request.GetForwardInfo().GetSourcePartition(),
78 > request.TaskQueue,
79 > )
80 >
81 > return c.client.AddWorkflowTask(ctx, request, opts...)
82 }
83
go.temporal.io/server/service/history/configs/quotas.go 13 introduced LOC · 1 range

Open complete file

37 return quotas.NewNamespaceRequestRateLimiter(
38 func(req quotas.Request) quotas.RequestRateLimiter {
39 > return quotas.NewPriorityRateLimiterHelper( quotas.go
40 > quotas.NewNamespaceRateBurst(
41 > req.Caller,
42 > namespaceRateFn,
43 > // TODO: We can consider adding a separate burst ratio dynamic config
44 > // on namespace level rate limiter if needed.
45 > quotas.DefaultIncomingNamespaceBurstRatioFn,
46 > ),
47 > operatorRPSRatio,
48 > RequestToPriority,
49 > APIPrioritiesOrdered,
50 > )
51 > },
52 )
53 }
go.temporal.io/server/client/matching/loadbalancer.go 10 introduced LOC · 5 ranges

Open complete file

79 taskQueue *tqid.TaskQueue,
80 pc PartitionCounts,
81 > ) *tqid.NormalPartition { loadbalancer.go
82 > if n, ok := testhooks.Get(lb.testHooks, testhooks.MatchingLBForceWritePartition, namespace.ID(taskQueue.NamespaceId())); ok {
83 return taskQueue.NormalPartition(n)
84 }
85
86 > nsName, err := lb.namespaceIDToName(namespace.ID(taskQueue.NamespaceId())) loadbalancer.go
87 > if err != nil {
88 return taskQueue.RootPartition()
89 }
90
91 > var partitionCount int loadbalancer.go
92 > if pc.Write > 0 {
93 partitionCount = int(pc.Write)
94 > } else { loadbalancer.go
95 > partitionCount = max(1, lb.nWritePartitions(nsName.String(), taskQueue.Name(), taskQueue.TaskType()))
96 > }
97
98 > return taskQueue.NormalPartition(rand.Intn(partitionCount)) loadbalancer.go
99 }
100
go.temporal.io/server/client/history/metric_client.go 9 introduced LOC · 3 ranges

Open complete file

88 ctx context.Context,
89 operation string,
90 > ) (metrics.Handler, time.Time) { metric_client.go
91 > caller := headers.GetCallerInfo(ctx).CallerName
92 > metricsHandler := c.metricsHandler.WithTags(metrics.OperationTag(operation), metrics.NamespaceTag(caller), metrics.ServiceRoleTag(metrics.HistoryRoleTagValue))
93 > metrics.ClientRequests.With(metricsHandler).Record(1)
94 > return metricsHandler, time.Now().UTC()
95 > }
96
97 func (c *metricClient) finishMetricsRecording(
99 startTime time.Time,
100 err error,
101 > ) { metric_client.go
102 > if err != nil {
103 switch err.(type) {
104 case *serviceerror.Canceled,
116 metrics.ClientFailures.With(metricsHandler).Record(1, metrics.ServiceErrorTypeTag(err))
117 }
118 > metrics.ClientLatency.With(metricsHandler).Record(time.Since(startTime)) metric_client.go
119 }

Direct hierarchy

Parents

Children