Vihren Atlas

mutable_state_impl.go ×22

Code frontier

unlabeled · c_0a1baf47b3dc

4 tests · 5080 LOC · 211 files · introduces 0 tests · 148 LOC · 3 files

Introduces — evidence that enters the hierarchy at this concept

Code
37 ranges148 lines · 3 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
832 ranges5080 lines · 211 files · Browse complete extent
All tests (intent)
4 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: 148 introduced LOC across 37 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/service/history/workflow/mutable_state_impl.go 73 introduced LOC · 22 ranges

Open complete file

9187 updatedSignalInfos map[int64]*persistencespb.SignalInfo,
9188 isSnapshot bool,
9189 > ) error { mutable_state_impl.go
9190 > err := applyUpdatesToSubStateMachine(ms, ms.pendingActivityInfoIDs, ms.updateActivityInfos, updatedActivityInfos, isSnapshot, ms.DeleteActivity, func(current, incoming *persistencespb.ActivityInfo) {
9191 if current == nil || ms.ShouldResetActivityTimerTaskMask(current, incoming) {
9192 incoming.TimerTaskStatus = TimerTaskStatusNone
9198 ms.activityInfosUserDataUpdated[ai.ScheduledEventId] = struct{}{}
9199 })
9200 > if err != nil { mutable_state_impl.go
9201 return err
9202 }
9203
9204 > err = applyUpdatesToSubStateMachine(ms, ms.pendingTimerInfoIDs, ms.updateTimerInfos, updatedTimerInfos, isSnapshot, ms.DeleteUserTimer, func(current, incoming *persistencespb.TimerInfo) { mutable_state_impl.go
9205 incoming.TaskStatus = TimerTaskStatusNone
9206 }, func(ti *persistencespb.TimerInfo) {
9208 ms.timerInfosUserDataUpdated[ti.TimerId] = struct{}{}
9209 })
9210 > if err != nil { mutable_state_impl.go
9211 return err
9212 }
9213
9214 > err = applyUpdatesToSubStateMachine(ms, ms.pendingChildExecutionInfoIDs, ms.updateChildExecutionInfos, updatedChildExecutionInfos, isSnapshot, ms.DeletePendingChildExecution, func(current, incoming *persistencespb.ChildExecutionInfo) { mutable_state_impl.go
9215 if current != nil {
9216 incoming.Clock = current.Clock
9217 }
9218 }, nil)
9219 > if err != nil { mutable_state_impl.go
9220 return err
9221 }
9222
9223 > err = applyUpdatesToSubStateMachine(ms, ms.pendingRequestCancelInfoIDs, ms.updateRequestCancelInfos, updatedRequestCancelInfos, isSnapshot, ms.DeletePendingRequestCancel, nil, nil) mutable_state_impl.go
9224 > if err != nil {
9225 return err
9226 }
9227
9228 > err = applyUpdatesToSubStateMachine(ms, ms.pendingSignalInfoIDs, ms.updateSignalInfos, updatedSignalInfos, isSnapshot, ms.DeletePendingSignal, nil, nil) mutable_state_impl.go
9229 > return err
9230 }
9231
9288 }
9289
9290 > func (ms *MutableStateImpl) applySignalRequestedIds(signalRequestedIds []string, incomingExecutionInfo *persistencespb.WorkflowExecutionInfo) { mutable_state_impl.go
9291 > // TODO(long-nt-tran): Deprecate this function once we fully ramp up writing signals to workflow CHASM component
9292 > if transitionhistory.Compare(
9293 > incomingExecutionInfo.SignalRequestIdsLastUpdateVersionedTransition,
9294 > ms.executionInfo.SignalRequestIdsLastUpdateVersionedTransition,
9295 > ) == 0 {
9296 > return
9297 > }
9298
9299 ids := make(map[string]struct{}, len(signalRequestedIds))
9323 sanitizeFn func(current, incoming V),
9324 postUpdateFn func(V),
9325 > ) error { mutable_state_impl.go
9326 > if isSnapshot {
9327 for key := range pendingInfos {
9328 if _, ok := updatedSubStateMachine[key]; !ok {
9335 }
9336
9337 > getSizeOfKey := func(key any) int { mutable_state_impl.go
9338 switch v := key.(type) {
9339 case string:
9344 }
9345
9346 > for key, updated := range updatedSubStateMachine { mutable_state_impl.go
9347 var existing V
9348 if existing, ok := pendingInfos[key]; ok {
9370 updatedUpdateInfos map[string]*persistencespb.UpdateInfo,
9371 isSnapshot bool,
9372 > ) { mutable_state_impl.go
9373 > if ms.executionInfo.UpdateInfos == nil {
9374 > ms.executionInfo.UpdateInfos = make(map[string]*persistencespb.UpdateInfo, len(updatedUpdateInfos))
9375 > }
9376 > if isSnapshot {
9377 for updateID := range ms.executionInfo.UpdateInfos {
9378 if _, ok := updatedUpdateInfos[updateID]; !ok {
9383 }
9384
9385 > for updateID, ui := range updatedUpdateInfos { mutable_state_impl.go
9386 if existing, ok := ms.executionInfo.UpdateInfos[updateID]; ok {
9387 if transitionhistory.Compare(existing.GetLastUpdateVersionedTransition(), ui.GetLastUpdateVersionedTransition()) == 0 {
9397 }
9398
9399 > func (ms *MutableStateImpl) syncExecutionInfo(current *persistencespb.WorkflowExecutionInfo, incoming *persistencespb.WorkflowExecutionInfo, isSnapshot bool) error { mutable_state_impl.go
9400 > var workflowTaskVersionUpdated bool
9401 > if transitionhistory.Compare(current.WorkflowTaskLastUpdateVersionedTransition, incoming.WorkflowTaskLastUpdateVersionedTransition) != 0 {
9402 ms.workflowTaskManager.UpdateWorkflowTask(&historyi.WorkflowTaskInfo{
9403 Version: incoming.WorkflowTaskVersion,
9424 }
9425
9426 > if incoming.WorkflowTaskType == enumsspb.WORKFLOW_TASK_TYPE_SPECULATIVE { mutable_state_impl.go
9427 ms.workflowTaskManager.deleteWorkflowTask()
9428 ms.RemoveSpeculativeWorkflowTaskTimeoutTask()
9435 }
9436
9437 > doNotSync := func(v any) []any { mutable_state_impl.go
9438 > info, ok := v.(*persistencespb.WorkflowExecutionInfo)
9439 > if !ok || info == nil {
9440 return nil
9441 }
9442 > ignoreFields := []any{ mutable_state_impl.go
9443 > &info.WorkflowTaskVersion,
9444 > &info.WorkflowTaskScheduledEventId,
9445 > &info.WorkflowTaskStartedEventId,
9446 > &info.WorkflowTaskRequestId,
9447 > &info.WorkflowTaskTimeout,
9448 > &info.WorkflowTaskAttempt,
9449 > &info.WorkflowTaskStartedTime,
9450 > &info.WorkflowTaskScheduledTime,
9451 > &info.WorkflowTaskOriginalScheduledTime,
9452 > &info.WorkflowTaskType,
9453 > &info.WorkflowTaskSuggestContinueAsNew,
9454 > &info.WorkflowTaskSuggestContinueAsNewReasons,
9455 > &info.WorkflowTaskHistorySizeBytes,
9456 > &info.WorkflowTaskBuildId,
9457 > &info.WorkflowTaskBuildIdRedirectCounter,
9458 > &info.VersionHistories,
9459 > &info.ExecutionStats,
9460 > &info.LastFirstEventTxnId,
9461 > &info.ParentClock,
9462 > &info.CloseTransferTaskId,
9463 > &info.CloseVisibilityTaskId,
9464 > &info.RelocatableAttributesRemoved,
9465 > &info.WorkflowExecutionTimerTaskStatus,
9466 > &info.SubStateMachinesByType,
9467 > &info.StateMachineTimers,
9468 > &info.TaskGenerationShardClockTimestamp,
9469 > &info.UpdateInfos,
9470 > }
9471 > if !isSnapshot {
9472 ignoreFields = append(ignoreFields, &info.SubStateMachineTombstoneBatches)
9473 }
9474 > return ignoreFields mutable_state_impl.go
9475 }
9476 > err := common.MergeProtoExcludingFields(current, incoming, doNotSync) mutable_state_impl.go
9477 > if err != nil {
9478 return err
9479 }
9480
9481 > ms.ClearStickyTaskQueue() mutable_state_impl.go
9482 >
9483 > return nil
9484 }
9485
go.temporal.io/server/service/history/ndc/workflow_state_replicator.go 65 introduced LOC · 13 ranges

Open complete file

488
489 // Separate logic based on whether there's a new run
490 > if versionedTransition.NewRunInfo == nil { workflow_state_replicator.go
491 > return r.handleFirstReplicationTaskWithoutNewRun(
492 > ctx,
493 > archetypeID,
494 > wfCtx,
495 > executionInfo,
496 > executionState,
497 > mutation,
498 > snapshot,
499 > versionedTransition,
500 > sourceClusterName,
501 > )
502 > }
503
504 return r.handleFirstReplicationTaskWithNewRun(
523 versionedTransition *replicationspb.VersionedTransitionArtifact,
524 sourceClusterName string,
525 > ) (continueProcess bool, retErr error) { workflow_state_replicator.go
526 > nsEntry, err := r.namespaceRegistry.GetNamespaceByID(namespace.ID(executionInfo.NamespaceId))
527 > if err != nil {
528 return false, err
529 }
530
531 > localMutableState := workflow.NewMutableState( workflow_state_replicator.go
532 > r.shardContext,
533 > r.shardContext.GetEventsCache(),
534 > r.logger,
535 > nsEntry,
536 > executionInfo.WorkflowId,
537 > executionState.RunId,
538 > timestamp.TimeValue(executionState.StartTime),
539 > )
540 > if archetypeID == chasm.WorkflowArchetypeID {
541 > err = localMutableState.SetHistoryTree(executionInfo.WorkflowExecutionTimeout, executionInfo.WorkflowRunTimeout, executionState.RunId)
542 > if err != nil {
543 return false, err
544 }
545 }
546
547 > newBranchToken, err := r.bringLocalEventsUpToSourceCurrentBranch( workflow_state_replicator.go
548 > ctx,
549 > namespace.ID(executionInfo.NamespaceId),
550 > executionInfo.WorkflowId,
551 > executionState.RunId,
552 > sourceClusterName,
553 > wfCtx,
554 > localMutableState,
555 > executionInfo.VersionHistories,
556 > versionedTransition.EventBatches,
557 > true,
558 > )
559 > if err != nil {
560 return false, err
561 }
562
563 > defer func() { workflow_state_replicator.go
564 > r.deleteNewBranchWhenError(ctx, namespace.ID(executionInfo.NamespaceId), executionInfo.WorkflowId, executionState.RunId, archetypeID, newBranchToken, retErr)
565 > }()
566
567 > if mutation != nil { workflow_state_replicator.go
568 err = localMutableState.ApplyMutation(mutation.StateMutation)
569 > } else { workflow_state_replicator.go
570 err = localMutableState.ApplySnapshot(snapshot.State)
571 }
572 > if err != nil { workflow_state_replicator.go
573 return false, err
574 }
575
576 > err = r.taskRefresher.Refresh(ctx, localMutableState, false) workflow_state_replicator.go
577 > if err != nil {
578 return false, err
579 }
580
581 > err = r.transactionMgr.CreateWorkflow( workflow_state_replicator.go
582 > ctx,
583 > archetypeID,
584 > NewWorkflow(
585 > r.clusterMetadata,
586 > wfCtx,
587 > localMutableState,
588 > wcache.NoopReleaseFn,
589 > ),
590 > )
591 > if errors.Is(err, consts.ErrDuplicate) {
592 return true, nil
593 }
594
595 > return false, err workflow_state_replicator.go
596 }
597
1249
1250 if versionhistory.IsEmptyVersionHistory(sourceVersionHistory) {
1251 > // we don't need to insert events but need to switch localMutableState to use workflow_state_replicator.go
1252 > // an empty version history if it is not already using one.
1253 > if !versionhistory.IsEmptyVersionHistory(localVersionHistory) {
1254 newIndex := versionhistory.AddEmptyVersionHistory(localVersionHistories)
1255 localVersionHistories.CurrentVersionHistoryIndex = newIndex
1256 }
1257 > return nil, nil workflow_state_replicator.go
1258 }
1259
go.temporal.io/server/service/history/ndc/transaction_manager_mock.go 10 introduced LOC · 2 ranges

Open complete file

79
80 // CreateWorkflow mocks base method.
81 > func (m *MockTransactionManager) CreateWorkflow(ctx context.Context, archetypeID chasm.ArchetypeID, targetWorkflow Workflow) error { transaction_manager_mock.go
82 > m.ctrl.T.Helper()
83 > ret := m.ctrl.Call(m, "CreateWorkflow", ctx, archetypeID, targetWorkflow)
84 > ret0, _ := ret[0].(error)
85 > return ret0
86 > }
87
88 // CreateWorkflow indicates an expected call of CreateWorkflow.
89 > func (mr *MockTransactionManagerMockRecorder) CreateWorkflow(ctx, archetypeID, targetWorkflow any) *gomock.Call { transaction_manager_mock.go
90 > mr.mock.ctrl.T.Helper()
91 > return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CreateWorkflow", reflect.TypeOf((*MockTransactionManager)(nil).CreateWorkflow), ctx, archetypeID, targetWorkflow)
92 > }
93
94 // GetCurrentWorkflowRunID mocks base method.

Direct hierarchy

Parents

Children