Vihren Atlas

api.go ×11

Code frontier

unlabeled · c_9f4d5c534ebc

3 tests · 9482 LOC · 250 files · introduces 0 tests · 140 LOC · 3 files

Introduces — evidence that enters the hierarchy at this concept

Code
18 ranges140 lines · 3 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
2056 ranges9482 lines · 250 files · Browse complete extent
All tests (intent)
3 testsBrowse complete intent

Neighbourhood graph

The orange circle is the focus. Violet and green circles are every ancestor and descendant, broader and narrower, at any distance; blue squares and pink diamonds are the introduced files and exact introduced tests of every visible concept, not only the focus's. Arrows point from broader to narrower concepts and bridge only concepts omitted from this view. Undirected links show source or test introduction. Concept and file size follows LOC; exact test nodes use test-count units.

Introduced files, introduced tests, and structurally relevant concept specialization

In the embedded map, ordinary wheel input scrolls the page; use the visible controls to zoom and drag to pan. Open the full-screen map for canvas navigation: wheel pans, Ctrl/Command plus wheel zooms, and arrow keys pan when this region is focused. On touch screens, open the full-screen map to pan or pinch. If JavaScript or WebGL is unavailable, use the native relationship evidence on this page.

Graph controls are ready.

Interactive rendering requires JavaScript and WebGL. Use the native relationship evidence on this page while the interactive map is unavailable.

Native relationship evidence

Every exact file and test below is linked only from the concept that introduces it.

Introduced tests

Every collected test enters the hierarchy at exactly one concept.

No tests are introduced at this concept. Its intent tests are introduced by other concepts.

Introduced code

Every collected source range enters the hierarchy at exactly one concept.

3 files ranked by introduced lines: 140 introduced LOC across 18 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/service/history/api/respondworkflowtaskcompleted/api.go 98 introduced LOC · 11 ranges

Open complete file

620 // skip transfer task for workflow task if request asking to return new workflow task
621 if bypassTaskGeneration {
622 > // start the new workflow task if request asked to do so api.go
623 > // TODO: replace the poll request
624 > _, newWorkflowTask, err = ms.AddWorkflowTaskStartedEvent(
625 > newWorkflowTask.ScheduledEventID,
626 > "request-from-RespondWorkflowTaskCompleted",
627 > newWorkflowTask.TaskQueue,
628 > request.Identity,
629 > versioningStamp,
630 > nil,
631 > workflowLease.GetContext().UpdateRegistry(ctx),
632 > false,
633 > nil,
634 > -1, // sentinel: inline path didn't consult matching, has no routing revision
635 > )
636 > if err != nil {
637 return nil, err
638 }
782 //nolint:staticcheck
783 if newWorkflowTask != nil && bypassTaskGeneration {
784 > resp.StartedResponse, err = recordworkflowtaskstarted.CreateRecordWorkflowTaskStartedResponse( api.go
785 > ctx,
786 > ms,
787 > updateRegistry,
788 > newWorkflowTask,
789 > request.GetIdentity(),
790 > request.GetForceCreateNewWorkflowTask(),
791 > )
792 > if err != nil {
793 return nil, err
794 }
795 // sticky is always enabled when worker request for new workflow task from RespondWorkflowTaskCompleted
796 > resp.StartedResponse.StickyExecutionEnabled = true api.go
797 >
798 > resp.NewWorkflowTask, err = handler.withNewWorkflowTask(ctx, namespaceEntry.Name(), req, resp.StartedResponse)
799 > if err != nil {
800 return nil, err
801 }
871 branchToken []byte,
872 maximumPageSize int32,
873 > ) (_ *workflowservice.PollWorkflowTaskQueueResponse, retError error) { api.go
874 >
875 > if matchingResp.WorkflowExecution == nil {
876 // this will happen if there is no workflow task to be send to worker / caller
877 return &workflowservice.PollWorkflowTaskQueueResponse{}, nil
878 }
879
880 > var history *historypb.History api.go
881 > var continuation []byte
882 > var err error
883 >
884 > if matchingResp.GetStickyExecutionEnabled() && matchingResp.Query != nil {
885 // meaning sticky query, we should not return any events to worker
886 // since query task only check the current status
888 Events: []*historypb.HistoryEvent{},
889 }
890 > } else { api.go
891 > // here we have 3 cases:
892 > // 1. sticky && non query task
893 > // 2. non sticky && non query task
894 > // 3. non sticky && query task
895 > // for 1, partial history have to be send back
896 > // for 2 and 3, full history have to be send back
897 >
898 > var persistenceToken []byte
899 >
900 > firstEventID := common.FirstEventID
901 > nextEventID := matchingResp.GetNextEventId()
902 > if matchingResp.GetStickyExecutionEnabled() {
903 > firstEventID = matchingResp.GetPreviousStartedEventId() + 1
904 > }
905
906 // TODO below is a temporal solution to guard against invalid event batch
907 // when data inconsistency occurs
908 // long term solution should check event batch pointing backwards within history store
909 > defer func() { api.go
910 > var dataLossErr *serviceerror.DataLoss
911 > if errors.As(retError, &dataLossErr) {
912 api.TrimHistoryNode(
913 ctx,
921 }
922 }()
923 > history, persistenceToken, err = api.GetHistory( api.go
924 > ctx,
925 > handler.shardContext,
926 > namespaceName,
927 > namespaceID,
928 > matchingResp.GetWorkflowExecution(),
929 > firstEventID,
930 > nextEventID,
931 > maximumPageSize,
932 > nil,
933 > matchingResp.GetTransientWorkflowTask(),
934 > branchToken,
935 > handler.persistenceVisibilityMgr,
936 > )
937 > if err != nil {
938 return nil, err
939 }
979 request *historyservice.RespondWorkflowTaskCompletedRequest,
980 response *historyservice.RecordWorkflowTaskStartedResponse,
981 > ) (*workflowservice.PollWorkflowTaskQueueResponse, error) { api.go
982 > taskToken, err := handler.tokenSerializer.Deserialize(request.CompleteRequest.TaskToken)
983 > if err != nil {
984 return nil, consts.ErrDeserializingToken
985 }
986
987 > taskToken = tasktoken.NewWorkflowTaskToken( api.go
988 > taskToken.GetNamespaceId(),
989 > taskToken.GetWorkflowId(),
990 > taskToken.GetRunId(),
991 > response.GetScheduledEventId(),
992 > response.GetStartedEventId(),
993 > response.GetStartedTime(),
994 > response.GetAttempt(),
995 > response.GetClock(),
996 > response.GetVersion(),
997 > )
998 > token, err := handler.tokenSerializer.Serialize(taskToken)
999 > if err != nil {
1000 return nil, err
1001 }
1002 > workflowExecution := &commonpb.WorkflowExecution{ api.go
1003 > WorkflowId: taskToken.GetWorkflowId(),
1004 > RunId: taskToken.GetRunId(),
1005 > }
1006 > matchingResp := common.CreateMatchingPollWorkflowTaskQueueResponse(response, workflowExecution, token)
1007 > return handler.createPollWorkflowTaskQueueResponse(
1008 > ctx,
1009 > namespaceName,
1010 > namespace.ID(taskToken.NamespaceId),
1011 > matchingResp,
1012 > matchingResp.GetBranchToken(),
1013 > int32(handler.config.HistoryMaxPageSize(namespaceName.String())),
1014 > )
1015 }
1016
go.temporal.io/server/service/history/api/recordworkflowtaskstarted/api.go 22 introduced LOC · 2 ranges

Open complete file

372 identity string,
373 wtHeartbeat bool,
374 > ) (*historyservice.RecordWorkflowTaskStartedResponse, error) { api.go
375 > rawResp, err := CreateRecordWorkflowTaskStartedResponseWithRawHistory(ctx, ms, updateRegistry, workflowTask, identity, wtHeartbeat)
376 > if err != nil {
377 return nil, err
378 }
379 > return &historyservice.RecordWorkflowTaskStartedResponse{ api.go
380 > WorkflowType: rawResp.WorkflowType,
381 > PreviousStartedEventId: rawResp.PreviousStartedEventId,
382 > ScheduledEventId: rawResp.ScheduledEventId,
383 > StartedEventId: rawResp.StartedEventId,
384 > NextEventId: rawResp.NextEventId,
385 > Attempt: rawResp.Attempt,
386 > StickyExecutionEnabled: rawResp.StickyExecutionEnabled,
387 > TransientWorkflowTask: rawResp.TransientWorkflowTask,
388 > WorkflowExecutionTaskQueue: rawResp.WorkflowExecutionTaskQueue,
389 > BranchToken: rawResp.BranchToken,
390 > ScheduledTime: rawResp.ScheduledTime,
391 > StartedTime: rawResp.StartedTime,
392 > Queries: rawResp.Queries,
393 > Clock: rawResp.Clock,
394 > Messages: rawResp.Messages,
395 > Version: rawResp.Version,
396 > NextPageToken: rawResp.NextPageToken,
397 > }, nil
398 }
399
go.temporal.io/server/api/matchingservice/v1/request_response.pb.go 20 introduced LOC · 5 ranges

Open complete file

431 }
432
433 > func (x *PollWorkflowTaskQueueResponseWithRawHistory) GetPreviousStartedEventId() int64 { request_response.pb.go
434 > if x != nil {
435 > return x.PreviousStartedEventId
436 > }
437 return 0
438 }
452 }
453
454 > func (x *PollWorkflowTaskQueueResponseWithRawHistory) GetNextEventId() int64 { request_response.pb.go
455 > if x != nil {
456 > return x.NextEventId
457 > }
458 return 0
459 }
466 }
467
468 > func (x *PollWorkflowTaskQueueResponseWithRawHistory) GetStickyExecutionEnabled() bool { request_response.pb.go
469 > if x != nil {
470 > return x.StickyExecutionEnabled
471 > }
472 return false
473 }
480 }
481
482 > func (x *PollWorkflowTaskQueueResponseWithRawHistory) GetTransientWorkflowTask() *v13.TransientWorkflowTaskInfo { request_response.pb.go
483 > if x != nil {
484 > return x.TransientWorkflowTask
485 > }
486 return nil
487 }
494 }
495
496 > func (x *PollWorkflowTaskQueueResponseWithRawHistory) GetBranchToken() []byte { request_response.pb.go
497 > if x != nil {
498 > return x.BranchToken
499 > }
500 return nil
501 }

Direct hierarchy

Parents

Children