Vihren Atlas

history_manager.go ×31

Code frontier

unlabeled · c_4be39e457337

353 tests · 3218 LOC · 155 files · introduces 0 tests · 156 LOC · 2 files

Introduces — evidence that enters the hierarchy at this concept

Code
32 ranges156 lines · 2 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
505 ranges3218 lines · 155 files · Browse complete extent
All tests (intent)
353 testsBrowse complete intent

Neighbourhood graph

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

Introduced files, introduced tests, and structurally relevant concept specialization

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

Graph controls are ready.

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

Native relationship evidence

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

Introduced tests

Every collected test enters the hierarchy at exactly one concept.

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

Introduced code

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

2 files ranked by introduced lines: 156 introduced LOC across 32 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/common/persistence/history_manager.go 145 introduced LOC · 31 ranges

Open complete file

616 pageSize int,
617 metadataOnly bool,
618 > ) ([]InternalHistoryNode, *historyPagingToken, error) { history_manager.go
619 >
620 > if token.CurrentRangeIndex == notStartedIndex {
621 > for idx, br := range branchAncestors {
622 > // this range won't contain any nodes needed
623 > if minNodeID >= br.GetEndNodeId() {
624 continue
625 }
626 // similarly, the ranges and the rest won't contain any nodes needed,
627 > if maxNodeID <= br.GetBeginNodeId() { history_manager.go
628 break
629 }
630
631 > if token.CurrentRangeIndex == notStartedIndex { history_manager.go
632 > token.CurrentRangeIndex = idx
633 > }
634 > token.FinalRangeIndex = idx
635 }
636
637 > if token.CurrentRangeIndex == notStartedIndex { history_manager.go
638 return nil, nil, softassert.UnexpectedDataLoss(m.logger, "branchRange is corrupted", nil)
639 }
640 }
641
642 > currentBranch := branchAncestors[token.CurrentRangeIndex] history_manager.go
643 > // minNodeID remains the same, since caller can read from the middle
644 > // maxNodeID need to be shortened since this branch can contain additional history nodes
645 > if currentBranch.GetEndNodeId() < maxNodeID {
646 maxNodeID = currentBranch.GetEndNodeId()
647 }
648 > branchID := currentBranch.GetBranchId() history_manager.go
649 > resp, err := m.persistence.ReadHistoryBranch(ctx, &InternalReadHistoryBranchRequest{
650 > BranchToken: branchToken,
651 > ShardID: shardID,
652 > BranchID: branchID,
653 > MinNodeID: minNodeID,
654 > MaxNodeID: maxNodeID,
655 > NextPageToken: token.StoreToken,
656 > PageSize: pageSize,
657 > MetadataOnly: metadataOnly,
658 > })
659 > if err != nil {
660 return nil, nil, err
661 }
662 > token.StoreToken = resp.NextPageToken history_manager.go
663 > return resp.Nodes, token, nil
664 }
665
739 return nil, nil, nil, nil, 0, serviceerror.NewInvalidArgument(fmt.Sprintf("unable to parse branch token: %v", err))
740 }
741 > branchID := branch.BranchId history_manager.go
742 > branchAncestors := branch.Ancestors
743 >
744 > // merge tree ID & branch ID into branch ancestors so the processing logic is simple
745 > beginNodeID := common.FirstEventID
746 > if len(branch.Ancestors) > 0 {
747 beginNodeID = branch.Ancestors[len(branch.Ancestors)-1].GetEndNodeId()
748 }
749 > branchAncestors = append(branchAncestors, &persistencespb.HistoryBranchRange{ history_manager.go
750 > BranchId: branchID,
751 > BeginNodeId: beginNodeID,
752 > EndNodeId: maxNodeID,
753 > })
754 >
755 > token, err := m.deserializeToken(
756 > request.NextPageToken,
757 > request.MinEventID-1,
758 > defaultLastTransactionID,
759 > )
760 > if err != nil {
761 return nil, nil, nil, nil, 0, err
762 }
763
764 > nodes, token, err := m.readRawHistoryBranch( history_manager.go
765 > ctx,
766 > branchToken,
767 > shardID,
768 > branchAncestors,
769 > minNodeID,
770 > maxNodeID,
771 > token,
772 > request.PageSize,
773 > false,
774 > )
775 > if err != nil {
776 return nil, nil, nil, nil, 0, err
777 }
778 > if len(nodes) == 0 && len(request.NextPageToken) == 0 { history_manager.go
779 return nil, nil, nil, nil, 0, serviceerror.NewNotFound("Workflow execution history not found.")
780 }
781
782 > nodes, err = m.filterHistoryNodes( history_manager.go
783 > token.LastNodeID,
784 > token.LastTransactionID,
785 > nodes,
786 > )
787 > if err != nil {
788 return nil, nil, nil, nil, 0, err
789 }
790
791 > var dataBlobs []*commonpb.DataBlob history_manager.go
792 > transactionIDs := make([]int64, 0, len(nodes))
793 > nodeIDs := make([]int64, 0, len(nodes))
794 > dataSize := 0
795 > if len(nodes) > 0 {
796 > dataBlobs = make([]*commonpb.DataBlob, len(nodes))
797 > for index, node := range nodes {
798 > dataBlobs[index] = node.Events
799 > if node.Events == nil {
800 return nil, nil, nil, nil, 0, softassert.UnexpectedDataLoss(m.logger, "no events in history node", nil)
801 }
802 > dataSize += len(node.Events.Data) history_manager.go
803 > transactionIDs = append(transactionIDs, node.TransactionID)
804 > nodeIDs = append(nodeIDs, node.NodeID)
805 }
806 > lastNode := nodes[len(nodes)-1] history_manager.go
807 > token.LastNodeID = lastNode.NodeID
808 > token.LastTransactionID = lastNode.TransactionID
809 }
810 > return dataBlobs, transactionIDs, nodeIDs, token, dataSize, nil history_manager.go
811 }
812
911 }
912
913 > historyEvents := make([]*historypb.HistoryEvent, 0, request.PageSize) history_manager.go
914 > historyEventBatches := make([]*historypb.History, 0, request.PageSize)
915 >
916 > var firstEvent, lastEvent *historypb.HistoryEvent
917 > var eventCount int
918 >
919 > dataLossTags := func(cause error) []tag.Tag {
920 return []tag.Tag{
921 tag.Cause(cause.Error()),
931 }
932
933 > for _, batch := range dataBlobs { history_manager.go
934 > events, err := m.serializer.DeserializeEvents(batch)
935 > if err != nil {
936 return nil, nil, nil, nil, dataSize, err
937 }
938 > if len(events) == 0 { history_manager.go
939 return nil, nil, nil, nil, dataSize, softassert.UnexpectedDataLoss(m.logger, dataLossMsg, errEmptyEvents, dataLossTags(errEmptyEvents)...)
940 }
941
942 > firstEvent = events[0] history_manager.go
943 > eventCount = len(events)
944 > lastEvent = events[eventCount-1]
945 >
946 > if firstEvent.GetVersion() != lastEvent.GetVersion() || firstEvent.GetEventId()+int64(eventCount-1) != lastEvent.GetEventId() {
947 // in a single batch, version should be the same, and ID should be contiguous
948 return historyEvents, historyEventBatches, transactionIDs, nil, dataSize, softassert.UnexpectedDataLoss(m.logger, dataLossMsg, errWrongVersion, dataLossTags(errWrongVersion)...)
949 }
950 > if firstEvent.GetEventId() != token.LastEventID+1 { history_manager.go
951 return historyEvents, historyEventBatches, transactionIDs, nil, dataSize, softassert.UnexpectedDataLoss(m.logger, dataLossMsg, errNonContiguousEventID, dataLossTags(errNonContiguousEventID)...)
952 }
953
954 > if byBatch { history_manager.go
955 historyEventBatches = append(historyEventBatches, &historypb.History{Events: events})
956 > } else { history_manager.go
957 > historyEvents = append(historyEvents, events...)
958 > }
959 > token.LastEventID = lastEvent.GetEventId()
960 }
961
962 > nextPageToken, err := m.serializeToken(token, false) history_manager.go
963 > if err != nil {
964 return nil, nil, nil, nil, 0, err
965 }
966 > return historyEvents, historyEventBatches, transactionIDs, nextPageToken, dataSize, nil history_manager.go
967 }
968
1040 lastTransactionID int64,
1041 nodes []InternalHistoryNode,
1042 > ) ([]InternalHistoryNode, error) { history_manager.go
1043 > var result []InternalHistoryNode
1044 > for _, node := range nodes {
1045 > // assuming that business logic layer is correct and transaction ID only increase
1046 > // thus, valid event batch will come with increasing transaction ID
1047 >
1048 > // event batches with smaller node ID
1049 > // -> should not be possible since records are already sorted
1050 > // event batches with same node ID
1051 > // -> batch with higher transaction ID is valid
1052 > // event batches with larger node ID
1053 > // -> batch with lower transaction ID is invalid (happens before)
1054 > // -> batch with higher transaction ID is valid
1055 > if node.TransactionID < lastTransactionID {
1056 continue
1057 }
1058
1059 > switch { history_manager.go
1060 case node.NodeID < lastNodeID:
1061 return nil, softassert.UnexpectedDataLoss(m.logger, "corrupted data, nodeID cannot decrease", nil)
1062 case node.NodeID == lastNodeID:
1063 return nil, softassert.UnexpectedDataLoss(m.logger, "corrupted data, same nodeID must have smaller txnID", nil)
1064 > default: // row.NodeID > lastNodeID: history_manager.go
1065 > // NOTE: when row.nodeID > lastNodeID, we expect the one with largest txnID comes first
1066 > lastTransactionID = node.TransactionID
1067 > lastNodeID = node.NodeID
1068 > result = append(result, node)
1069 }
1070 }
1071 > return result, nil history_manager.go
1072 }
1073
1106 defaultLastEventID int64,
1107 lastTransactionId int64,
1108 > ) (*historyPagingToken, error) { history_manager.go
1109 >
1110 > return m.pagingTokenSerializer.Deserialize(
1111 > token,
1112 > defaultLastEventID,
1113 > defaultLastNodeID,
1114 > lastTransactionId,
1115 > )
1116 > }
1117
1118 func (m *executionManagerImpl) serializeToken(
1119 pagingToken *historyPagingToken,
1120 reverseOrder bool,
1121 > ) ([]byte, error) { history_manager.go
1122 >
1123 > if len(pagingToken.StoreToken) == 0 {
1124 > if pagingToken.CurrentRangeIndex == pagingToken.FinalRangeIndex {
1125 > // this means that we have reached the final page of final branchRange
1126 > return nil, nil
1127 > }
1128
1129 if reverseOrder {
go.temporal.io/server/common/persistence/json_history_token_serializer.go 11 introduced LOC · 1 range

Open complete file

49 defaultLastNodeID int64,
50 defaultLastTransactionID int64,
51 > ) (*historyPagingToken, error) { json_history_token_serializer.go
52 >
53 > if len(data) == 0 {
54 > token := historyPagingToken{
55 > LastEventID: defaultLastEventID,
56 > CurrentRangeIndex: notStartedIndex,
57 > LastNodeID: defaultLastNodeID,
58 > LastTransactionID: defaultLastTransactionID,
59 > }
60 > return &token, nil
61 > }
62
63 token := historyPagingToken{}

Direct hierarchy

Parents

Children