Vihren Atlas

mutable_state_impl.go ×20

Code frontier

unlabeled · c_079fe05133b1

7 tests · 9743 LOC · 260 files · introduces 0 tests · 150 LOC · 3 files

Introduces — evidence that enters the hierarchy at this concept

Code
27 ranges150 lines · 3 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
2161 ranges9743 lines · 260 files · Browse complete extent
All tests (intent)
7 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: 150 introduced LOC across 27 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/service/history/workflow/mutable_state_impl.go 79 introduced LOC · 20 ranges

Open complete file

3350 requestID string,
3351 updateCallbacks []*commonpb.Callback,
3352 > ) error { mutable_state_impl.go
3353 > if len(updateCallbacks) == 0 {
3354 > return nil
3355 > }
3356 if ms.chasmCallbacksEnabled() && ms.config.EnableWorkflowUpdateCallbacks(ms.GetNamespaceEntry().Name().String()) {
3357 // Initialize chasm tree once for new workflows.
5647 acceptedRequestSequencingEventID int64,
5648 acceptedRequest *updatepb.Request,
5649 > ) (*historypb.HistoryEvent, error) { mutable_state_impl.go
5650 > if err := ms.checkMutability(tag.WorkflowActionUpdateAccepted); err != nil {
5651 return nil, err
5652 }
5653 > event := ms.hBuilder.AddWorkflowExecutionUpdateAcceptedEvent(updateID, acceptedRequestMessageID, acceptedRequestSequencingEventID, acceptedRequest) mutable_state_impl.go
5654 > if err := ms.ApplyWorkflowExecutionUpdateAcceptedEvent(event); err != nil {
5655 return nil, err
5656 }
5657 // Add links from Nexus callbacks to the event.
5658 > callbacksLinks := make([]*commonpb.Link, 0) mutable_state_impl.go
5659 > for _, cb := range acceptedRequest.GetCompletionCallbacks() {
5660 if cb.GetNexus() != nil {
5661 callbacksLinks = append(callbacksLinks, cb.GetLinks()...)
5662 }
5663 }
5664 > event.Links = callbacksLinks mutable_state_impl.go
5665 > return event, nil
5666 }
5667
5668 func (ms *MutableStateImpl) ApplyWorkflowExecutionUpdateAcceptedEvent(
5669 event *historypb.HistoryEvent,
5670 > ) error { mutable_state_impl.go
5671 > attrs := event.GetWorkflowExecutionUpdateAcceptedEventAttributes()
5672 > if attrs == nil {
5673 return serviceerror.NewInternal("wrong event type in call to ApplyWorkflowExecutionUpdateAcceptedEvent")
5674 }
5675 > if ms.executionInfo.UpdateInfos == nil { mutable_state_impl.go
5676 > ms.executionInfo.UpdateInfos = make(map[string]*persistencespb.UpdateInfo, 1)
5677 > }
5678 // NOTE: `attrs.GetAcceptedRequest().GetMeta().GetUpdateId()` was used here before, but that is problematic
5679 // in a reset/conflict resolution scenario where there is no `acceptedRequest` since the previously written
5680 // UpdateAdmitted event already contains the Update payload.
5681 > updateID := attrs.GetProtocolInstanceId() mutable_state_impl.go
5682 > var sizeDelta int
5683 > if ui, ok := ms.executionInfo.UpdateInfos[updateID]; ok {
5684 sizeBefore := ui.Size()
5685 ui.Value = &persistencespb.UpdateInfo_Acceptance{
5687 }
5688 sizeDelta = ui.Size() - sizeBefore
5689 > } else { mutable_state_impl.go
5690 > ui := persistencespb.UpdateInfo{
5691 > Value: &persistencespb.UpdateInfo_Acceptance{
5692 > Acceptance: &persistencespb.UpdateAcceptanceInfo{EventId: event.EventId},
5693 > },
5694 > }
5695 > ms.executionInfo.UpdateInfos[updateID] = &ui
5696 > ms.executionInfo.UpdateCount++
5697 > sizeDelta = ui.Size() + len(updateID)
5698 > }
5699 > ms.approximateSize += sizeDelta
5700 > ms.updateInfoUpdated[updateID] = struct{}{}
5701 > ms.writeEventToCache(event)
5702 > // Add update completion callbacks.
5703 > // This is the primary path for registering callbacks — AcceptedRequest is
5704 > // present in the normal flow. The exception is the reset/reapply case where
5705 > // callbacks are registered at admission time instead (because the
5706 > // UpdateAccepted event has a nil AcceptedRequest after reset). In that case,
5707 > // addCallbacksToMap is a no-op since the requestID-indexed keys already
5708 > // exist from the admitted event.
5709 > if attrs.GetAcceptedRequest() != nil {
5710 > requestID := attrs.GetAcceptedRequest().GetRequestId()
5711 > if requestID != "" {
5712 ms.AttachRequestID(requestID, event.EventType, event.EventId)
5713 }
5714 > if err := ms.addUpdateCallbacks( mutable_state_impl.go
5715 > event,
5716 > updateID,
5717 > requestID,
5718 > attrs.GetAcceptedRequest().GetCompletionCallbacks(),
5719 > ); err != nil {
5720 return err
5721 }
5722 }
5723 > return nil mutable_state_impl.go
5724 }
5725
5727 acceptedEventID int64,
5728 updResp *updatepb.Response,
5729 > ) (*historypb.HistoryEvent, error) { mutable_state_impl.go
5730 > if err := ms.checkMutability(tag.WorkflowActionUpdateCompleted); err != nil {
5731 return nil, err
5732 }
5733 > event, batchID := ms.hBuilder.AddWorkflowExecutionUpdateCompletedEvent(acceptedEventID, updResp) mutable_state_impl.go
5734 > if err := ms.ApplyWorkflowExecutionUpdateCompletedEvent(event, batchID); err != nil {
5735 return nil, err
5736 }
5737 > return event, nil mutable_state_impl.go
5738 }
5739
5741 event *historypb.HistoryEvent,
5742 batchID int64,
5743 > ) error { mutable_state_impl.go
5744 > attrs := event.GetWorkflowExecutionUpdateCompletedEventAttributes()
5745 > if attrs == nil {
5746 return serviceerror.NewInternal("wrong event type in call to ApplyWorkflowExecutionUpdateCompletedEvent")
5747 }
5748 > if ms.executionInfo.UpdateInfos == nil { mutable_state_impl.go
5749 // UpdateInfo must be created by preceding UpdateAccepted event.
5750 return serviceerror.NewInvalidArgument("WorkflowExecutionUpdateCompletedEvent doesn't have preceding WorkflowExecutionUpdateAcceptedEvent")
5751 }
5752 > updateID := attrs.GetMeta().GetUpdateId() mutable_state_impl.go
5753 > var sizeDelta int
5754 > ui, uiExists := ms.executionInfo.UpdateInfos[updateID]
5755 > if !uiExists {
5756 // UpdateInfo must be created by preceding UpdateAccepted event.
5757 return serviceerror.NewInvalidArgument("WorkflowExecutionUpdateCompletedEvent doesn't have preceding WorkflowExecutionUpdateAcceptedEvent")
5758 }
5759 > sizeBefore := ui.Size() mutable_state_impl.go
5760 > ui.Value = &persistencespb.UpdateInfo_Completion{
5761 > Completion: &persistencespb.UpdateCompletionInfo{
5762 > EventId: event.EventId,
5763 > EventBatchId: batchID,
5764 > },
5765 > }
5766 > sizeDelta = ui.Size() - sizeBefore
5767 > ms.approximateSize += sizeDelta
5768 > ms.updateInfoUpdated[updateID] = struct{}{}
5769 > if ms.ChasmEnabled() {
5770 if err := ms.processUpdateCallbacks(updateID); err != nil {
5771 return err
5772 }
5773 }
5774 > ms.writeEventToCache(event) mutable_state_impl.go
5775 > return nil
5776 }
5777
7927
7928 for updateID := range ms.updateInfoUpdated {
7929 > ms.executionInfo.UpdateInfos[updateID].LastUpdateVersionedTransition = currentVersionedTransition mutable_state_impl.go
7930 > }
7931
7932 if ms.workflowTaskUpdated {
go.temporal.io/server/common/testing/updateutils/update.go 68 introduced LOC · 6 ranges

Open complete file

29 }
30
31 > func (u UpdateUtils) UpdateAcceptCommands(tv *testvars.TestVars) []*commandpb.Command { update.go
32 > if th, ok := u.t.(helper); ok {
33 > th.Helper()
34 > }
35
36 > return []*commandpb.Command{{ update.go
37 > CommandType: enumspb.COMMAND_TYPE_PROTOCOL_MESSAGE,
38 > Attributes: &commandpb.Command_ProtocolMessageCommandAttributes{ProtocolMessageCommandAttributes: &commandpb.ProtocolMessageCommandAttributes{
39 > MessageId: tv.MessageID() + "_update-accepted",
40 > }},
41 > }}
42 }
43
44 > func (u UpdateUtils) UpdateCompleteCommands(tv *testvars.TestVars) []*commandpb.Command { update.go
45 > if th, ok := u.t.(helper); ok {
46 > th.Helper()
47 > }
48 > return []*commandpb.Command{
49 > {
50 > CommandType: enumspb.COMMAND_TYPE_PROTOCOL_MESSAGE,
51 > Attributes: &commandpb.Command_ProtocolMessageCommandAttributes{ProtocolMessageCommandAttributes: &commandpb.ProtocolMessageCommandAttributes{
52 > MessageId: tv.MessageID() + "_update-completed",
53 > }},
54 > },
55 > }
56 }
57
63 }
64
65 > func (u UpdateUtils) UpdateAcceptMessages(tv *testvars.TestVars, updRequestMsg *protocolpb.Message) []*protocolpb.Message { update.go
66 > if th, ok := u.t.(helper); ok {
67 > th.Helper()
68 > }
69 > updRequest := protoutils.UnmarshalAny[*updatepb.Request](u.t, updRequestMsg.GetBody())
70 >
71 > return []*protocolpb.Message{
72 > {
73 > Id: tv.MessageID() + "_update-accepted",
74 > ProtocolInstanceId: updRequest.GetMeta().GetUpdateId(),
75 > SequencingId: nil,
76 > Body: protoutils.MarshalAny(u.t, &updatepb.Acceptance{
77 > AcceptedRequestMessageId: updRequestMsg.GetId(),
78 > AcceptedRequestSequencingEventId: updRequestMsg.GetEventId(),
79 > AcceptedRequest: updRequest,
80 > }),
81 > },
82 > }
83 }
84
85 > func (u UpdateUtils) UpdateCompleteMessages(tv *testvars.TestVars, updRequestMsg *protocolpb.Message) []*protocolpb.Message { update.go
86 > if th, ok := u.t.(helper); ok {
87 > th.Helper()
88 > }
89 > updRequest := protoutils.UnmarshalAny[*updatepb.Request](u.t, updRequestMsg.GetBody())
90 >
91 > return []*protocolpb.Message{
92 > {
93 > Id: tv.MessageID() + "_update-completed",
94 > ProtocolInstanceId: updRequest.GetMeta().GetUpdateId(),
95 > SequencingId: nil,
96 > Body: protoutils.MarshalAny(u.t, &updatepb.Response{
97 > Meta: updRequest.GetMeta(),
98 > Outcome: &updatepb.Outcome{
99 > Value: &updatepb.Outcome_Success{
100 > Success: payloads.EncodeString("success-result-of-" + updRequest.GetMeta().GetUpdateId()),
101 > },
102 > },
103 > }),
104 > },
105 > }
106 }
107
108 > func (u UpdateUtils) UpdateAcceptCompleteMessages(tv *testvars.TestVars, updRequestMsg *protocolpb.Message) []*protocolpb.Message { update.go
109 > if th, ok := u.t.(helper); ok {
110 > th.Helper()
111 > }
112 > return append(
113 > u.UpdateAcceptMessages(tv, updRequestMsg),
114 > u.UpdateCompleteMessages(tv, updRequestMsg)...)
115 }
116
go.temporal.io/server/api/persistence/v1/update.go-helpers.pb.go 3 introduced LOC · 1 range

Open complete file

128
129 // Size returns the size of the object, in bytes, once serialized
130 > func (val *UpdateInfo) Size() int { update.go-helpers.pb.go
131 > return proto.Size(val)
132 > }
133
134 // Equal returns whether two UpdateInfo values are equivalent by recursively

Direct hierarchy

Parents

Children