Vihren Atlas

api.go ×18

Joint frontier

unlabeled · c_887c78acddb9

1 test · 6511 LOC · 237 files · introduces 1 test · 122 LOC · 5 files

Introduces — evidence that enters the hierarchy at this concept

Code
27 ranges122 lines · 5 files
Tests
1 test

Contains — complete concept membership

All code (extent)
1191 ranges6511 lines · 237 files · Browse complete extent
All tests (intent)
1 testBrowse 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.

1 test introduced at this concept.

Introduced code

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

5 files ranked by introduced lines: 122 introduced LOC across 27 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/service/history/api/getworkflowexecutionhistory/api.go 87 introduced LOC · 18 ranges

Open complete file

47 cachedTransientTasks *historyspb.TransientWorkflowTaskInfo,
48 nextEventID int64,
49 > ) { api.go
50 > if !shardContext.GetConfig().SendTransientOrSpeculativeWorkflowTaskEvents(namespaceName) {
51 return
52 }
53
54 // CLI and UI clients should not receive transient events for backward compatibility
55 > if !api.ClientSupportsTranOrSpecEvents(ctx) { api.go
56 return
57 }
58
59 > var transientWorkflowTask *historyspb.TransientWorkflowTaskInfo api.go
60 >
61 > // Try cached tasks first
62 > if cachedTransientTasks != nil {
63 // Validate cached tasks are still valid (not stale)
64 if err := api.ValidateTransientWorkflowTaskEvents(nextEventID, cachedTransientTasks); err == nil {
68
69 // If no valid cache, fetch fresh from mutable state
70 > if transientWorkflowTask == nil { api.go
71 > msResp, err := api.GetOrPollWorkflowMutableState(
72 > ctx,
73 > shardContext,
74 > &historyservice.GetMutableStateRequest{
75 > NamespaceId: namespaceID.String(),
76 > Execution: execution,
77 > },
78 > workflowConsistencyChecker,
79 > eventNotifier,
80 > )
81 > if err != nil {
82 // Transient events don't exist or are already committed - this is OK
83 // Just return without appending (events are in persisted history)
84 return
85 }
86 > transientWorkflowTask = msResp.GetTransientOrSpeculativeTasks() api.go
87 > if transientWorkflowTask == nil {
88 return
89 }
90
91 > if msResp.GetWorkflowStatus() != enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING { api.go
92 return
93 }
94
95 > if err := api.ValidateTransientWorkflowTaskEvents(nextEventID, transientWorkflowTask); err != nil { api.go
96 return
97 }
99
100 // Manually append transient events to the response
101 > if useRawHistory { api.go
102 > transientEventsBlob, err := shardContext.GetPayloadSerializer().SerializeEvents(transientWorkflowTask.GetHistorySuffix())
103 > if err == nil {
104 > *historyBlob = append(*historyBlob, transientEventsBlob)
105 > } else {
106 softassert.Fail(shardContext.GetLogger(), "GetWorkflowExecutionHistory could not serialize transient workflow tasks")
107 shardContext.GetLogger().Error("Failed to serialize transient workflow tasks",
231 queryNextEventID := common.EndEventID
232 if request.Request.NextPageToken != nil {
233 > continuationToken, err = api.DeserializeHistoryToken(request.Request.NextPageToken) api.go
234 > if err != nil {
235 return nil, consts.ErrInvalidNextPageToken
236 }
237 > if execution.GetRunId() != "" && execution.GetRunId() != continuationToken.GetRunId() { api.go
238 return nil, consts.ErrNextPageTokenRunIDMismatch
239 }
240
241 > execution.RunId = continuationToken.GetRunId() api.go
242 >
243 > // Set isWorkflowRunning from continuation token for pagination
244 > isWorkflowRunning = continuationToken.IsWorkflowRunning
245 >
246 > // we need to update the current next event ID and whether workflow is running
247 > if len(continuationToken.PersistenceToken) == 0 && isLongPoll && continuationToken.IsWorkflowRunning {
248 if !isCloseEventOnly {
249 queryNextEventID = continuationToken.GetNextEventId()
397 continuationToken = nil
398 }
399 > } else { api.go
400 > // return all events
401 > if continuationToken.FirstEventId >= continuationToken.NextEventId {
402 // currently there is no new event
403 history.Events = []*historypb.HistoryEvent{}
405 continuationToken = nil
406 }
407 > } else { api.go
408 > if sendRawWorkflowHistoryForNamespace || sendRawHistoryBetweenInternalServices {
409 > historyBlob, continuationToken.PersistenceToken, err = api.GetRawHistory(
410 > ctx,
411 > shardContext,
412 > namespaceName,
413 > namespaceID,
414 > execution,
415 > continuationToken.FirstEventId,
416 > continuationToken.NextEventId,
417 > request.Request.GetMaximumPageSize(),
418 > continuationToken.PersistenceToken,
419 > nil,
420 > continuationToken.BranchToken,
421 > )
422 > } else {
423 history, continuationToken.PersistenceToken, err = api.GetHistory(
424 ctx,
437 }
438
439 > if err != nil { api.go
440 return nil, err
441 }
442
443 // Query and append transient/speculative tasks if on last page
444 > if len(continuationToken.PersistenceToken) == 0 { api.go
445 > // Re-query mutable state to detect events committed to DB during pagination (race condition fix).
446 > // When a speculative/transient WFT times out or fails between the first and last DB page fetches,
447 > // those events are committed to DB with IDs < continuationToken.NextEventId but were excluded
448 > // because the DB fetch was capped at the original boundary. Fetch the gap now, then update
449 > // the nextEventID boundary so appendTransientTasks validates against the correct ID.
450 > _, _, _, freshNextEventID, freshIsRunning, freshVersionHistoryItem, freshVersionedTransition, freshTransientTasks, freshErr :=
451 > queryMutableState(namespaceID, execution, common.EmptyEventID,
452 > continuationToken.BranchToken, continuationToken.VersionHistoryItem, continuationToken.VersionedTransition)
453 > if freshErr != nil {
454 return nil, freshErr
455 }
456 > if freshNextEventID > continuationToken.NextEventId { api.go
457 // Events were committed to DB during pagination — fetch the gap.
458 if freshErr = fetchGapEvents(continuationToken.NextEventId, freshNextEventID, continuationToken.BranchToken); freshErr != nil {
468 }
469
470 > appendTransientTasks( api.go
471 > ctx,
472 > shardContext,
473 > workflowConsistencyChecker,
474 > eventNotifier,
475 > namespaceID,
476 > namespaceName.String(),
477 > execution,
478 > sendRawWorkflowHistoryForNamespace || sendRawHistoryBetweenInternalServices,
479 > history,
480 > &historyBlob,
481 > cachedTransientTasks,
482 > continuationToken.NextEventId,
483 > )
484 }
485
486 // here, for long pull on history events, we need to intercept the paging token from cassandra
487 // and do something clever
488 > if len(continuationToken.PersistenceToken) == 0 && (!continuationToken.IsWorkflowRunning || !isLongPoll) { api.go
489 > // meaning, there is no more history to be returned
490 > continuationToken = nil
491 > }
492 }
493 }
go.temporal.io/server/api/token/v1/message.pb.go 18 introduced LOC · 3 ranges

Open complete file

42 }
43
44 > func (x *HistoryContinuation) Reset() { message.pb.go
45 > *x = HistoryContinuation{}
46 > mi := &file_temporal_server_api_token_v1_message_proto_msgTypes[0]
47 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
48 > ms.StoreMessageInfo(mi)
49 > }
50
51 func (x *HistoryContinuation) String() string {
55 func (*HistoryContinuation) ProtoMessage() {}
56
57 > func (x *HistoryContinuation) ProtoReflect() protoreflect.Message { message.pb.go
58 > mi := &file_temporal_server_api_token_v1_message_proto_msgTypes[0]
59 > if x != nil {
60 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
61 > if ms.LoadMessageInfo() == nil {
62 > ms.StoreMessageInfo(mi)
63 > }
64 > return ms
65 }
66 return mi.MessageOf(x)
72 }
73
74 > func (x *HistoryContinuation) GetRunId() string { message.pb.go
75 > if x != nil {
76 > return x.RunId
77 > }
78 return ""
79 }
go.temporal.io/server/api/token/v1/message.go-helpers.pb.go 6 introduced LOC · 2 ranges

Open complete file

7
8 // Marshal an object of type HistoryContinuation to the protobuf v3 wire format
9 > func (val *HistoryContinuation) Marshal() ([]byte, error) { message.go-helpers.pb.go
10 > return proto.Marshal(val)
11 > }
12
13 // Unmarshal an object of type HistoryContinuation from the protobuf v3 wire format
14 > func (val *HistoryContinuation) Unmarshal(buf []byte) error { message.go-helpers.pb.go
15 > return proto.Unmarshal(buf, val)
16 > }
17
18 // Size returns the size of the object, in bytes, once serialized
go.temporal.io/server/service/history/api/token.go 6 introduced LOC · 2 ranges

Open complete file

69 return nil, nil
70 }
71 > return token.Marshal() token.go
72 }
73
74 // NOTE: DO NOT MODIFY UNLESS ALSO APPLIED TO ./service/frontend/token_deprecated.go
75 > func DeserializeHistoryToken(bytes []byte) (*tokenspb.HistoryContinuation, error) { token.go
76 > token := &tokenspb.HistoryContinuation{}
77 > err := token.Unmarshal(bytes)
78 > return token, err
79 > }
80
81 func GeneratePaginationToken(
go.temporal.io/server/service/history/api/get_history_util.go 5 introduced LOC · 2 ranges

Open complete file

441 eventIDOffset int64,
442 transientWorkflowTaskInfo *historyspb.TransientWorkflowTaskInfo,
443 > ) error { get_history_util.go
444 > for i, event := range transientWorkflowTaskInfo.HistorySuffix {
445 > expectedEventID := eventIDOffset + int64(i)
446 > if event.GetEventId() != expectedEventID {
447 return serviceerror.NewInternalf(
448 "invalid transient workflow task at position %v; expected event ID %v, found event ID %v",

Direct hierarchy

Children