Vihren Atlas

mutable_state_impl.go ×30

Code frontier

unlabeled · c_67bd6464828c

271 tests · 4182 LOC · 165 files · introduces 0 tests · 90 LOC · 1 file

Introduces — evidence that enters the hierarchy at this concept

Code
30 ranges90 lines · 1 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
681 ranges4182 lines · 165 files · Browse complete extent
All tests (intent)
271 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.

1 file ranked by introduced lines: 90 introduced LOC across 30 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/service/history/workflow/mutable_state_impl.go 90 introduced LOC · 30 ranges

Open complete file

7730 }
7731
7732 > for nodePath := range chasmNodesMutation.DeletedNodes { mutable_state_impl.go
7733 ms.approximateSize -= ms.chasmNodeSizes[nodePath]
7734 delete(ms.chasmNodeSizes, nodePath)
7735 }
7736 > for nodePath, node := range chasmNodesMutation.UpdatedNodes { mutable_state_impl.go
7737 newSize := len(nodePath) + node.Size()
7738 ms.approximateSize += newSize - ms.chasmNodeSizes[nodePath]
7740 }
7741
7742 > if isStateDirty { mutable_state_impl.go
7743 if err := ms.closeTransactionUpdateTransitionHistory(
7744 transactionPolicy,
7751
7752 // todo@TimeSkipping, we can move update versioned transition to inside closeTransactionHandleWorkflowTimeSkipping
7753 > ms.closeTransactionTrackLastUpdateVersionedTransition( mutable_state_impl.go
7754 > transactionPolicy,
7755 > )
7756 >
7757 > ms.closeTransactionTrackTombstones(transactionPolicy, chasmNodesMutation)
7758 >
7759 > // generate tasks
7760 > if err := ms.closeTransactionPrepareTasks(
7761 > transactionPolicy,
7762 > eventBatches,
7763 > clearBuffer,
7764 > regenTimerTasksForWorkflowTimeSkipping,
7765 > ); err != nil {
7766 return closeTransactionResult{}, err
7767 }
7768
7769 > ms.executionInfo.StateTransitionCount += 1 mutable_state_impl.go
7770 > ms.executionInfo.LastUpdateTime = timestamppb.New(ms.timeSource.Now())
7771 >
7772 > // We generate checksum here based on the assumption that the returned
7773 > // snapshot object is considered immutable. As of this writing, the only
7774 > // code that modifies the returned object lives inside Context.resetWorkflowExecution.
7775 > // Currently, the updates done inside Context.resetWorkflowExecution don't
7776 > // impact the checksum calculation.
7777 > checksum := ms.generateChecksum()
7778 >
7779 > if ms.dbRecordVersion == 0 {
7780 // noop, existing behavior
7781 > } else { mutable_state_impl.go
7782 ms.dbRecordVersion += 1
7783 }
7784
7785 > return closeTransactionResult{ mutable_state_impl.go
7786 > workflowEventsSeq: workflowEventsSeq,
7787 > bufferEvents: bufferEvents,
7788 > clearBuffer: clearBuffer,
7789 > checksum: checksum,
7790 > chasmNodesMutation: chasmNodesMutation,
7791 > }, nil
7792 }
7793
7886 func (ms *MutableStateImpl) closeTransactionTrackLastUpdateVersionedTransition(
7887 transactionPolicy historyi.TransactionPolicy,
7888 > ) { mutable_state_impl.go
7889 > if transactionPolicy != historyi.TransactionPolicyActive {
7890 // TODO: replication/standby logic will need a different way for updating LastUpdatedVersionedTransition
7891 // when reapplying history, especially when history replication tasks got batched.
8044 transactionPolicy historyi.TransactionPolicy,
8045 chasmNodesMutation chasm.NodesMutation,
8046 > ) { mutable_state_impl.go
8047 > if transactionPolicy != historyi.TransactionPolicyActive {
8048 // Passive/Replication logic will update tombstone list when applying mutable state
8049 // snapshot or mutation.
8173 clearBufferEvents bool,
8174 regenerateTimerTasksForTimeSkipping bool,
8175 > ) error { mutable_state_impl.go
8176 > if err := ms.closeTransactionHandleWorkflowResetTask(
8177 > transactionPolicy,
8178 > ); err != nil {
8179 return err
8180 }
8181
8182 > if err := ms.taskGenerator.GenerateDirtySubStateMachineTasks(ms.shard.StateMachineRegistry()); err != nil { mutable_state_impl.go
8183 return err
8184 }
8185
8186 > ms.closeTransactionCollapseVisibilityTasks() mutable_state_impl.go
8187 >
8188 > if err := ms.closeTransactionGenerateChasmRetentionTask(transactionPolicy); err != nil {
8189 return err
8190 }
8195 // regardless of how many activity & user timer created
8196 // so the calculation must be at the very end
8197 > if err := ms.closeTransactionHandleActivityUserTimerTasks(transactionPolicy); err != nil { mutable_state_impl.go
8198 return err
8199 }
8200 > if regenerateTimerTasksForTimeSkipping { mutable_state_impl.go
8201 if err := ms.closeTransactionRegenTimerTasksForWorkflowTimeSkipping(transactionPolicy); err != nil {
8202 return err
8204 }
8205
8206 > return ms.closeTransactionPrepareReplicationTasks(transactionPolicy, eventBatches, clearBufferEvents) mutable_state_impl.go
8207 }
8208
8209 func (ms *MutableStateImpl) closeTransactionGenerateChasmRetentionTask(
8210 transactionPolicy historyi.TransactionPolicy,
8211 > ) error { mutable_state_impl.go
8212 >
8213 > if ms.IsWorkflow() ||
8214 > ms.executionState.State != enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED ||
8215 > ms.stateInDB == enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED {
8216 return nil
8217 }
8234 eventBatches [][]*historypb.HistoryEvent,
8235 clearBufferEvents bool,
8236 > ) error { mutable_state_impl.go
8237 > var replicationTasks []tasks.Task
8238 > if ms.config.ReplicationMultipleBatches() {
8239 task, err := ms.eventsToReplicationTask(transactionPolicy, eventBatches)
8240 if err != nil {
8242 }
8243 replicationTasks = append(replicationTasks, task...)
8244 > } else { mutable_state_impl.go
8245 > for _, historyEvents := range eventBatches {
8246 task, err := ms.eventsToReplicationTask(transactionPolicy, [][]*historypb.HistoryEvent{historyEvents})
8247 if err != nil {
8251 }
8252 }
8253 > replicationTasks = append(replicationTasks, ms.syncActivityToReplicationTask(transactionPolicy)...) mutable_state_impl.go
8254 > replicationTasks = append(replicationTasks, ms.dirtyHSMToReplicationTask(transactionPolicy, eventBatches, clearBufferEvents)...)
8255 >
8256 > archetypeID := ms.ChasmTree().ArchetypeID()
8257 > isWorkflow := archetypeID == chasm.WorkflowArchetypeID
8258 > if !isWorkflow && len(replicationTasks) != 0 {
8259 return softassert.UnexpectedInternalErr(ms.logger, "chasm execution generated workflow replication tasks", nil)
8260 }
8261
8262 > if ms.transitionHistoryEnabled { mutable_state_impl.go
8263 switch transactionPolicy {
8264 case historyi.TransactionPolicyActive:
8336 }
8337
8338 > if transactionPolicy == historyi.TransactionPolicyPassive && mutable_state_impl.go
8339 > len(ms.InsertTasks[tasks.CategoryReplication]) > 0 {
8340 return softassert.UnexpectedInternalErr(
8341 ms.logger,
8499 func (ms *MutableStateImpl) syncActivityToReplicationTask(
8500 transactionPolicy historyi.TransactionPolicy,
8501 > ) []tasks.Task { mutable_state_impl.go
8502 > now := time.Now().UTC()
8503 > switch transactionPolicy {
8504 case historyi.TransactionPolicyActive:
8505 if ms.generateReplicationTask() {
8537 eventBatches [][]*historypb.HistoryEvent,
8538 clearBufferEvents bool,
8539 > ) []tasks.Task { mutable_state_impl.go
8540 > switch transactionPolicy {
8541 case historyi.TransactionPolicyActive:
8542 if !ms.generateReplicationTask() {
8870 func (ms *MutableStateImpl) closeTransactionHandleWorkflowResetTask(
8871 transactionPolicy historyi.TransactionPolicy,
8872 > ) error { mutable_state_impl.go
8873 > if transactionPolicy == historyi.TransactionPolicyPassive ||
8874 > !ms.IsWorkflowExecutionRunning() {
8875 return nil
8876 }
8902 func (ms *MutableStateImpl) closeTransactionHandleActivityUserTimerTasks(
8903 transactionPolicy historyi.TransactionPolicy,
8904 > ) error { mutable_state_impl.go
8905 > switch transactionPolicy {
8906 case historyi.TransactionPolicyActive:
8907 if !ms.IsWorkflowExecutionRunning() {
8923 // Any other task type is preserved in order.
8924 // Eg: [START, UPSERT, TP1, CLOSE, TP2, TP3] -> [TP1, CLOSE, TP2, TP3]
8925 > func (ms *MutableStateImpl) closeTransactionCollapseVisibilityTasks() { mutable_state_impl.go
8926 > visTasks := ms.InsertTasks[tasks.CategoryVisibility]
8927 > if len(visTasks) < 2 {
8928 return
8929 }
8982 }
8983
8984 > func (ms *MutableStateImpl) generateChecksum() *persistencespb.Checksum { mutable_state_impl.go
8985 > if !ms.shouldGenerateChecksum() {
8986 > return nil
8987 > }
8988 csum, err := generateMutableStateChecksum(ms)
8989 if err != nil {
8994 }
8995
8996 > func (ms *MutableStateImpl) shouldGenerateChecksum() bool { mutable_state_impl.go
8997 > if ms.namespaceEntry == nil {
8998 return false
8999 }
9000 > return rand.Intn(100) < ms.config.MutableStateChecksumGenProbability(ms.namespaceEntry.Name().String()) mutable_state_impl.go
9001 }
9002

Direct hierarchy

Parents

Children