Vihren Atlas

get_history_util.go ×9

Code frontier

unlabeled · c_ee2dbb48de59

5 tests · 5764 LOC · 225 files · introduces 0 tests · 76 LOC · 3 files

Introduces — evidence that enters the hierarchy at this concept

Code
16 ranges76 lines · 3 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
1005 ranges5764 lines · 225 files · Browse complete extent
All tests (intent)
5 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: 76 introduced LOC across 16 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/service/history/api/get_history_util.go 50 introduced LOC · 9 ranges

Open complete file

41 transientWorkflowTaskInfo *historyspb.TransientWorkflowTaskInfo,
42 branchToken []byte,
43 > ) (_ []*commonpb.DataBlob, _ []byte, retError error) { get_history_util.go
44 > defer func() {
45 > var dataLossErr *serviceerror.DataLoss
46 > if errors.As(retError, &dataLossErr) {
47 if shardContext.GetConfig().EnableDataLossMetrics() {
48 persistence.EmitDataLossMetric(
58 }()
59
60 > logger := shardContext.GetLogger() get_history_util.go
61 > rawHistory, size, nextToken, err := persistence.ReadFullPageRawEvents(
62 > ctx, shardContext.GetExecutionManager(),
63 > &persistence.ReadHistoryBranchRequest{
64 > BranchToken: branchToken,
65 > MinEventID: firstEventID,
66 > MaxEventID: nextEventID,
67 > PageSize: int(pageSize),
68 > NextPageToken: token,
69 > ShardID: shardContext.GetShardID(),
70 > },
71 > )
72 >
73 > if err != nil {
74 return nil, nil, err
75 }
76
77 > allEvents := make([]*historyspb.StrippedHistoryEvent, 0) get_history_util.go
78 > var lastEventID int64
79 > for _, blob := range rawHistory {
80 > events, err := shardContext.GetPayloadSerializer().DeserializeStrippedEvents(blob)
81 > if err != nil {
82 return nil, nil, err
83 }
84 > err = persistence.ValidateBatch(events, branchToken, lastEventID, logger) get_history_util.go
85 > if err != nil {
86 return nil, nil, err
87 }
88 > allEvents = append(allEvents, events...) get_history_util.go
89 > lastEventID = events[len(events)-1].GetEventId()
90 }
91 > var firstEvent, lastEvent *historyspb.StrippedHistoryEvent get_history_util.go
92 > if len(allEvents) > 0 {
93 > firstEvent = allEvents[0]
94 > lastEvent = allEvents[len(allEvents)-1]
95 > }
96 > if err = VerifyHistoryIsComplete(
97 > logger,
98 > firstEvent,
99 > lastEvent,
100 > len(allEvents),
101 > firstEventID,
102 > nextEventID-1,
103 > len(token) == 0,
104 > len(nextToken) == 0,
105 > int(pageSize),
106 > ); err != nil {
107 metricsHandler := interceptor.GetMetricsHandlerFromContext(ctx, logger).WithTags(metrics.OperationTag(metrics.HistoryGetHistoryScope))
108 metrics.ServiceErrIncompleteHistoryCounter.With(metricsHandler).Record(1)
112 }
113
114 > metricsHandler := interceptor.GetMetricsHandlerFromContext(ctx, shardContext.GetLogger()).WithTags(metrics.OperationTag(metrics.HistoryGetHistoryScope)) get_history_util.go
115 > metrics.HistorySize.With(metricsHandler).Record(int64(size))
116 >
117 > if len(nextToken) == 0 && transientWorkflowTaskInfo != nil {
118 // Check if we should include transient/speculative events
119 if shouldIncludeTransientOrSpeculativeTasks(ctx, transientWorkflowTaskInfo) {
136 // Ensure all raw history is proto3 encoded since data may be stored in other formats during testing.
137 // In production (proto3 encoding), this returns the input unchanged.
138 > rawHistory, err = serialization.ReencodeEventBlobsAsProto3(shardContext.GetPayloadSerializer(), rawHistory) get_history_util.go
139 > if err != nil {
140 return nil, nil, err
141 }
142
143 > return rawHistory, nextToken, nil get_history_util.go
144 }
145
go.temporal.io/server/common/persistence/history_manager_util.go 22 introduced LOC · 6 ranges

Open complete file

45 executionMgr ExecutionManager,
46 req *ReadHistoryBranchRequest,
47 > ) ([]*commonpb.DataBlob, int, []byte, error) { history_manager_util.go
48 > var blobs []*commonpb.DataBlob
49 > size := 0
50 > for {
51 > response, err := executionMgr.ReadRawHistoryBranch(ctx, req)
52 > if err != nil {
53 return nil, 0, nil, err
54 }
55 > blobs = append(blobs, response.HistoryEventBlobs...) history_manager_util.go
56 > size += response.Size
57 > if len(blobs) >= req.PageSize || len(response.NextPageToken) == 0 {
58 > return blobs, size, response.NextPageToken, nil
59 > }
60 req.NextPageToken = response.NextPageToken
61 }
140 lastEventID int64,
141 logger log.Logger,
142 > ) error { history_manager_util.go
143 > var firstEvent, lastEvent *historyspb.StrippedHistoryEvent
144 > var eventCount int
145 > dataLossTags := func(cause error) []tag.Tag {
146 return []tag.Tag{
147 tag.Cause(cause.Error()),
155 }
156 }
157 > firstEvent = batch[0] history_manager_util.go
158 > eventCount = len(batch)
159 > lastEvent = batch[eventCount-1]
160 >
161 > if firstEvent.GetVersion() != lastEvent.GetVersion() || firstEvent.GetEventId()+int64(eventCount-1) != lastEvent.GetEventId() {
162 // in a single batch, version should be the same, and ID should be contiguous
163 return softassert.UnexpectedDataLoss(logger, dataLossMsg, errWrongVersion, dataLossTags(errWrongVersion)...)
165 // If it is the first batch in the response, we cannot check the first event id here. That information is in the historyPagingToken.
166 // TODO: PPV refactor to move this check to ExecutionManager so that we can include that check as well.
167 > if lastEventID != 0 && firstEvent.GetEventId() != lastEventID+1 { history_manager_util.go
168 return softassert.UnexpectedDataLoss(logger, dataLossMsg, errNonContiguousEventID, dataLossTags(errNonContiguousEventID)...)
169 }
170 > return nil history_manager_util.go
171 }
go.temporal.io/server/api/history/v1/message.pb.go 4 introduced LOC · 1 range

Open complete file

379 }
380
381 > func (x *StrippedHistoryEvent) GetVersion() int64 { message.pb.go
382 > if x != nil {
383 > return x.Version
384 > }
385 return 0
386 }

Direct hierarchy

Parents

Children