TestProcessCloseExecution_NoParent

298 runID string,

299 startTime time.Time,

300 > ) *MutableStateImpl { mutable_state_impl.go

301 >

302 > namespaceName := namespaceEntry.Name().String()

303 > logger = log.NewLazyLogger(logger, func() []tag.Tag {

304 > return []tag.Tag{ mutable_state_impl.go

305 > tag.WorkflowNamespace(namespaceName),

306 > tag.WorkflowID(workflowID),

307 > tag.WorkflowRunID(runID),

308 > }

309 > })

310

311 > s := &MutableStateImpl{ mutable_state_impl.go

312 > updateActivityInfos: make(map[int64]*persistencespb.ActivityInfo),

313 > pendingActivityTimerHeartbeats: make(map[int64]time.Time),

314 > pendingActivityInfoIDs: make(map[int64]*persistencespb.ActivityInfo),

315 > pendingActivityIDToEventID: make(map[string]int64),

316 > deleteActivityInfos: make(map[int64]struct{}),

317 > syncActivityTasks: make(map[int64]struct{}),

318 >

319 > pendingTimerInfoIDs: make(map[string]*persistencespb.TimerInfo),

320 > pendingTimerEventIDToID: make(map[int64]string),

321 > updateTimerInfos: make(map[string]*persistencespb.TimerInfo),

322 > deleteTimerInfos: make(map[string]struct{}),

323 >

324 > updateChildExecutionInfos: make(map[int64]*persistencespb.ChildExecutionInfo),

325 > pendingChildExecutionInfoIDs: make(map[int64]*persistencespb.ChildExecutionInfo),

326 > deleteChildExecutionInfos: make(map[int64]struct{}),

327 >

328 > updateRequestCancelInfos: make(map[int64]*persistencespb.RequestCancelInfo),

329 > pendingRequestCancelInfoIDs: make(map[int64]*persistencespb.RequestCancelInfo),

330 > deleteRequestCancelInfos: make(map[int64]struct{}),

331 >

332 > updateSignalInfos: make(map[int64]*persistencespb.SignalInfo),

333 > pendingSignalInfoIDs: make(map[int64]*persistencespb.SignalInfo),

334 > deleteSignalInfos: make(map[int64]struct{}),

335 >

336 > updateSignalRequestedIDs: make(map[string]struct{}),

337 > pendingSignalRequestedIDs: make(map[string]struct{}),

338 > deleteSignalRequestedIDs: make(map[string]struct{}),

339 >

340 > // This field will be initialized with a real chasm tree at the end of this function

341 > // when feature flag is enabled.

342 > chasmTree: &noopChasmTree{},

343 >

344 > approximateSize: 0,

345 > chasmNodeSizes: make(map[string]int),

346 > totalTombstones: 0,

347 > currentVersion: namespaceEntry.FailoverVersion(workflowID),

348 > bufferEventsInDB: nil,

349 > stateInDB: enumsspb.WORKFLOW_EXECUTION_STATE_VOID,

350 > nextEventIDInDB: common.FirstEventID,

351 > dbRecordVersion: 1,

352 > namespaceEntry: namespaceEntry,

353 > appliedEvents: make(map[string]struct{}),

354 > InsertTasks: make(map[tasks.Category][]tasks.Task),

355 > BestEffortDeleteTasks: make(map[tasks.Category][]tasks.Key),

356 > transitionHistoryEnabled: shard.GetConfig().EnableTransitionHistory(namespaceName),

357 > visibilityUpdated: false,

358 > executionStateUpdated: false,

359 > workflowTaskUpdated: false,

360 > updateInfoUpdated: make(map[string]struct{}),

361 > timerInfosUserDataUpdated: make(map[string]struct{}),

362 > activityInfosUserDataUpdated: make(map[int64]struct{}),

363 > reapplyEventsCandidate: []*historypb.HistoryEvent{},

364 >

365 > QueryRegistry: NewQueryRegistry(),

366 >

367 > shard: shard,

368 > clusterMetadata: shard.GetClusterMetadata(),

369 > eventsCache: eventsCache,

370 > config: shard.GetConfig(),

371 > timeSource: shard.GetTimeSource(),

372 > logger: logger,

373 > metricsHandler: shard.GetMetricsHandler().WithTags(metrics.OperationTag(metrics.WorkflowContextScope)),

374 > endpointRegistry: shard.EndpointRegistry(),

375 > }

376 >

377 > s.executionInfo = &persistencespb.WorkflowExecutionInfo{

378 > NamespaceId: namespaceEntry.ID().String(),

379 > WorkflowId: workflowID,

380 >

381 > WorkflowTaskVersion: common.EmptyVersion,

382 > WorkflowTaskScheduledEventId: common.EmptyEventID,

383 > WorkflowTaskStartedEventId: common.EmptyEventID,

384 > WorkflowTaskRequestId: emptyUUID,

385 > WorkflowTaskTimeout: timestamp.DurationFromSeconds(0),

386 > WorkflowTaskAttempt: 1,

387 >

388 > LastCompletedWorkflowTaskStartedEventId: common.EmptyEventID,

389 >

390 > StartTime: timestamppb.New(startTime),

391 > ExecutionTime: timestamppb.New(startTime),

392 > VersionHistories: versionhistory.NewVersionHistories(&historyspb.VersionHistory{}),

393 > ExecutionStats: &persistencespb.ExecutionStats{HistorySize: 0},

394 > SubStateMachinesByType: make(map[string]*persistencespb.StateMachineMap),

395 > }

396 > s.executionInfo.TaskGenerationShardClockTimestamp = shard.CurrentVectorClock().GetClock()

397 > s.approximateSize += s.executionInfo.Size()

398 >

399 > s.executionState = &persistencespb.WorkflowExecutionState{

400 > RunId: runID,

401 >

402 > State: enumsspb.WORKFLOW_EXECUTION_STATE_CREATED,

403 > Status: enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING,

404 > StartTime: timestamppb.New(startTime),

405 > RequestIds: make(map[string]*persistencespb.RequestIDInfo),

406 > }

407 > s.approximateSize += s.executionState.Size()

408 >

409 > s.hBuilder = historybuilder.New(

410 > s.timeSource,

411 > s.shard.GenerateTaskIDs,

412 > s.currentVersion,

413 > common.FirstEventID,

414 > s.bufferEventsInDB,

415 > s.metricsHandler,

416 > s.config.MaximumEventBatchSizeInBytes,

417 > )

418 > s.taskGenerator = GetTaskGeneratorProvider().NewTaskGenerator(shard, s)

419 > s.workflowTaskManager = newWorkflowTaskStateMachine(s, s.metricsHandler)

420 >

421 > s.mustInitHSM()

422 >

423 > // TODO@time-skipping: support time skipping for chasm

424 > if s.config.EnableChasm(namespaceName) {

425 s.chasmTree = chasm.NewEmptyTree(

426 shard.ChasmRegistry(),

433 }

434

435 > if s.executionInfo.GetTimeSkippingInfo() != nil { mutable_state_impl.go

436 s.wrapTimeSourceWithTimeSkipping()

437 }

438 > return s mutable_state_impl.go

439 }

440

446 dbRecord *persistencespb.WorkflowMutableState,

447 dbRecordVersion int64,

448 > ) (*MutableStateImpl, error) { mutable_state_impl.go

449 > // startTime will be overridden by DB record

450 > startTime := time.Time{}

451 >

452 > mutableState := NewMutableState(

453 > shard,

454 > eventsCache,

455 > logger,

456 > namespaceEntry,

457 > dbRecord.ExecutionInfo.WorkflowId,

458 > dbRecord.ExecutionState.RunId,

459 > startTime,

460 > )

461 >

462 > if dbRecord.ActivityInfos != nil {

463 mutableState.pendingActivityInfoIDs = dbRecord.ActivityInfos

464 mutableState.approximateSize += int64SizeBytes * len(mutableState.pendingActivityInfoIDs)

465 }

466 > for _, activityInfo := range dbRecord.ActivityInfos { mutable_state_impl.go

467 mutableState.pendingActivityIDToEventID[activityInfo.ActivityId] = activityInfo.ScheduledEventId

468 mutableState.approximateSize += activityInfo.Size()

474 }

475

476 > if dbRecord.TimerInfos != nil { mutable_state_impl.go

477 mutableState.pendingTimerInfoIDs = dbRecord.TimerInfos

478 }

479 > for timerID, timerInfo := range dbRecord.TimerInfos { mutable_state_impl.go

480 mutableState.pendingTimerEventIDToID[timerInfo.GetStartedEventId()] = timerInfo.GetTimerId()

481 mutableState.approximateSize += timerInfo.Size()

483 }

484

485 > if dbRecord.ChildExecutionInfos != nil { mutable_state_impl.go

486 mutableState.pendingChildExecutionInfoIDs = dbRecord.ChildExecutionInfos

487 mutableState.approximateSize += int64SizeBytes * len(mutableState.pendingChildExecutionInfoIDs)

488 }

489 > for _, childInfo := range dbRecord.ChildExecutionInfos { mutable_state_impl.go

490 mutableState.approximateSize += childInfo.Size()

491 }

492

493 > if dbRecord.RequestCancelInfos != nil { mutable_state_impl.go

494 mutableState.pendingRequestCancelInfoIDs = dbRecord.RequestCancelInfos

495 mutableState.approximateSize += int64SizeBytes * len(mutableState.pendingRequestCancelInfoIDs)

496 }

497 > for _, cancelInfo := range dbRecord.RequestCancelInfos { mutable_state_impl.go

498 mutableState.approximateSize += cancelInfo.Size()

499 }

500

501 > if dbRecord.SignalInfos != nil { mutable_state_impl.go

502 mutableState.pendingSignalInfoIDs = dbRecord.SignalInfos

503 mutableState.approximateSize += int64SizeBytes * len(mutableState.pendingSignalInfoIDs)

504 }

505 > for _, signalInfo := range dbRecord.SignalInfos { mutable_state_impl.go

506 mutableState.approximateSize += signalInfo.Size()

507 }

508

509 > mutableState.pendingSignalRequestedIDs = convert.StringSliceToSet(dbRecord.SignalRequestedIds) mutable_state_impl.go

510 > for requestID := range mutableState.pendingSignalRequestedIDs {

511 mutableState.approximateSize += len(requestID)

512 }

513

514 > for _, tombstoneBatch := range dbRecord.ExecutionInfo.SubStateMachineTombstoneBatches { mutable_state_impl.go

515 mutableState.totalTombstones += len(tombstoneBatch.StateMachineTombstones)

516 }

519 // conflict path can surface it without loading ExecutionInfo. Backfill in memory for records

520 // written before that change so the next persist writes it through.

521 > if dbRecord.ExecutionState.FirstExecutionRunId == "" && dbRecord.ExecutionInfo.FirstExecutionRunId != "" { mutable_state_impl.go

522 dbRecord.ExecutionState.FirstExecutionRunId = dbRecord.ExecutionInfo.FirstExecutionRunId

523 }

524

525 > mutableState.approximateSize += dbRecord.ExecutionState.Size() - mutableState.executionState.Size() mutable_state_impl.go

526 > mutableState.executionState = dbRecord.ExecutionState

527 > mutableState.approximateSize += dbRecord.ExecutionInfo.Size() - mutableState.executionInfo.Size()

528 > mutableState.executionInfo = dbRecord.ExecutionInfo

529 >

530 > // StartTime was moved from ExecutionInfo to executionState

531 > if mutableState.executionState.StartTime == nil && dbRecord.ExecutionInfo.StartTime != nil {

532 mutableState.executionState.StartTime = dbRecord.ExecutionInfo.StartTime

533 }

534

535 > mutableState.hBuilder = historybuilder.New( mutable_state_impl.go

536 > mutableState.timeSource,

537 > mutableState.shard.GenerateTaskIDs,

538 > common.EmptyVersion,

539 > dbRecord.NextEventId,

540 > dbRecord.BufferedEvents,

541 > mutableState.metricsHandler,

542 > mutableState.config.MaximumEventBatchSizeInBytes,

543 > )

544 > mutableState.currentVersion = common.EmptyVersion

545 > mutableState.bufferEventsInDB = dbRecord.BufferedEvents

546 > mutableState.stateInDB = dbRecord.ExecutionState.State

547 > mutableState.nextEventIDInDB = dbRecord.NextEventId

548 > mutableState.dbRecordVersion = dbRecordVersion

549 > mutableState.checksum = dbRecord.Checksum

550 > mutableState.initVersionedTransitionInDB()

551 >

552 > if len(dbRecord.Checksum.GetValue()) > 0 {

553 switch {

554 case mutableState.shouldInvalidateCheckum():

566 }

567

568 > mutableState.mustInitHSM() mutable_state_impl.go

569 >

570 > // Track chasm node size even if chasm is not enabled,

571 > // because those nodes are still stored in the mutable state,

572 > // and should be taken into account when deciding if execution

573 > // should be terminated based on mutable state size.

574 > for key, node := range dbRecord.ChasmNodes {

575 nodeSize := len(key) + node.Size()

576 mutableState.approximateSize += nodeSize

579

580 // TODO@time-skipping: support time skipping for chasm

581 > if shard.GetConfig().EnableChasm(namespaceEntry.Name().String()) { mutable_state_impl.go

582 var err error

583 mutableState.chasmTree, err = chasm.NewTreeFromDB(

594 }

595 }

596 > if mutableState.executionInfo.GetTimeSkippingInfo() != nil { mutable_state_impl.go

597 mutableState.wrapTimeSourceWithTimeSkipping()

598 }

599 > return mutableState, nil mutable_state_impl.go

600 }

601

655 }

656

657 > func (ms *MutableStateImpl) mustInitHSM() { mutable_state_impl.go

658 > if ms.executionInfo.SubStateMachinesByType == nil {

659 > ms.executionInfo.SubStateMachinesByType = make(map[string]*persistencespb.StateMachineMap) mutable_state_impl.go

660 > }

661

662 // Error only occurs if some initialization path forgets to register the workflow state machine.

663 > stateMachineNode, err := hsm.NewRoot(ms.shard.StateMachineRegistry(), StateMachineType, ms, ms.executionInfo.SubStateMachinesByType, ms) mutable_state_impl.go

664 > if err != nil {

665 panic(err)

666 }

667 > ms.stateMachineNode = stateMachineNode mutable_state_impl.go

668 }

669

670 > func (ms *MutableStateImpl) IsWorkflow() bool { mutable_state_impl.go

671 > return ms.chasmTree.ArchetypeID() == chasm.WorkflowArchetypeID

672 > }

673

674 > func (ms *MutableStateImpl) HSM() *hsm.Node { mutable_state_impl.go

675 > return ms.stateMachineNode

676 > }

677

678 > func (ms *MutableStateImpl) ChasmTree() historyi.ChasmTree { mutable_state_impl.go

679 > return ms.chasmTree

680 > }

681

682 // ChasmEnabled returns true if the mutable state has a real chasm tree.

684 // enabled when the mutable state is created. Once the EnableChasm dynamic config is removed and the tree is always

685 // initialized, this helper can be removed.

686 > func (ms *MutableStateImpl) ChasmEnabled() bool { mutable_state_impl.go

687 > _, isNoop := ms.chasmTree.(*noopChasmTree)

688 > return !isNoop

689 > }

690

691 // chasmCallbacksEnabled returns true if CHASM callbacks are enabled for this workflow.

972 }

973

974 > func (ms *MutableStateImpl) CloneToProto() *persistencespb.WorkflowMutableState { mutable_state_impl.go

975 > msProto := &persistencespb.WorkflowMutableState{

976 > ActivityInfos: ms.pendingActivityInfoIDs,

977 > TimerInfos: ms.pendingTimerInfoIDs,

978 > ChildExecutionInfos: ms.pendingChildExecutionInfoIDs,

979 > RequestCancelInfos: ms.pendingRequestCancelInfoIDs,

980 > SignalInfos: ms.pendingSignalInfoIDs,

981 > ChasmNodes: ms.chasmTree.Snapshot(nil).Nodes,

982 > SignalRequestedIds: convert.StringSetToSlice(ms.pendingSignalRequestedIDs),

983 > ExecutionInfo: ms.executionInfo,

984 > ExecutionState: ms.executionState,

985 > NextEventId: ms.hBuilder.NextEventID(),

986 > BufferedEvents: ms.bufferEventsInDB,

987 > Checksum: ms.checksum,

988 > }

989 >

990 > return common.CloneProto(msProto)

991 > }

992

993 > func (ms *MutableStateImpl) GetWorkflowKey() definition.WorkflowKey { mutable_state_impl.go

994 > return definition.NewWorkflowKey(

995 > ms.executionInfo.NamespaceId,

996 > ms.executionInfo.WorkflowId,

997 > ms.executionState.RunId,

998 > )

999 > }

1000

1001 > func (ms *MutableStateImpl) GetCurrentBranchToken() ([]byte, error) { mutable_state_impl.go

1002 > currentVersionHistory, err := versionhistory.GetCurrentVersionHistory(ms.executionInfo.VersionHistories)

1003 > if err != nil {

1004 return nil, err

1005 }

1006 > return currentVersionHistory.GetBranchToken(), nil mutable_state_impl.go

1007 }

1008

1024 runTimeout *durationpb.Duration,

1025 treeID string,

1026 > ) error { mutable_state_impl.go

1027 > // NOTE: Unfortunately execution timeout and run timeout are not yet initialized into ms.executionInfo at this point.

1028 > // TODO: Consider explicitly initializing mutable state with these timeout parameters instead of passing them in.

1029 > workflowKey := ms.GetWorkflowKey()

1030 >

1031 > archetypeID := ms.ChasmTree().ArchetypeID()

1032 > if archetypeID != chasm.WorkflowArchetypeID {

1033 return softassert.UnexpectedInternalErr(

1034 ms.logger,

1042 }

1043

1044 > var retentionDuration *durationpb.Duration mutable_state_impl.go

1045 > if duration := ms.namespaceEntry.Retention(); duration > 0 {

1046 > retentionDuration = durationpb.New(duration) mutable_state_impl.go

1047 > }

1048 > initialBranchToken, err := ms.shard.GetExecutionManager().GetHistoryBranchUtil().NewHistoryBranch( mutable_state_impl.go

1049 > workflowKey.NamespaceID,

1050 > workflowKey.WorkflowID,

1051 > workflowKey.RunID,

1052 > treeID,

1053 > nil,

1054 > []*persistencespb.HistoryBranchRange{},

1055 > runTimeout.AsDuration(),

1056 > executionTimeout.AsDuration(),

1057 > retentionDuration.AsDuration(),

1058 > )

1059 > if err != nil {

1060 return err

1061 }

1062 > return ms.SetCurrentBranchToken(initialBranchToken) mutable_state_impl.go

1063 }

1064

1065 func (ms *MutableStateImpl) SetCurrentBranchToken(

1066 branchToken []byte,

1067 > ) error { mutable_state_impl.go

1068 > currentVersionHistory, err := versionhistory.GetCurrentVersionHistory(ms.executionInfo.VersionHistories)

1069 > if err != nil {

1070 return err

1071 }

1072 > versionhistory.SetVersionHistoryBranchToken(currentVersionHistory, branchToken) mutable_state_impl.go

1073 > return nil

1074 }

1075

1120 }

1121

1122 > func (ms *MutableStateImpl) GetExecutionInfo() *persistencespb.WorkflowExecutionInfo { mutable_state_impl.go

1123 > return ms.executionInfo

1124 > }

1125

1126 > func (ms *MutableStateImpl) GetExecutionState() *persistencespb.WorkflowExecutionState { mutable_state_impl.go

1127 > return ms.executionState

1128 > }

1129

1130 > func (ms *MutableStateImpl) FlushBufferedEvents() { mutable_state_impl.go

1131 > if ms.HasStartedWorkflowTask() {

1132 return

1133 }

1134 > ms.updatePendingEventIDs(ms.hBuilder.FlushBufferToCurrentBatch()) mutable_state_impl.go

1135 }

1136

1138 version int64,

1139 forceUpdate bool,

1140 > ) error { mutable_state_impl.go

1141 > if ms.transitionHistoryEnabled && len(ms.executionInfo.TransitionHistory) != 0 {

1142 > // this make sure current version >= last write version mutable_state_impl.go

1143 > lastVersionedTransition := ms.CurrentVersionedTransition()

1144 > ms.currentVersion = lastVersionedTransition.NamespaceFailoverVersion

1145 > } else { mutable_state_impl.go

1146 versionHistory, err := versionhistory.GetCurrentVersionHistory(ms.executionInfo.VersionHistories)

1147 if err != nil {

1159 }

1160

1161 > if version > ms.currentVersion || forceUpdate { mutable_state_impl.go

1162 ms.currentVersion = version

1163 }

1164

1165 > ms.hBuilder = historybuilder.New( mutable_state_impl.go

1166 > ms.timeSource,

1167 > ms.shard.GenerateTaskIDs,

1168 > ms.currentVersion,

1169 > ms.nextEventIDInDB,

1170 > ms.bufferEventsInDB,

1171 > ms.metricsHandler,

1172 > ms.config.MaximumEventBatchSizeInBytes,

1173 > )

1174 >

1175 > return nil

1176 }

1177

1178 > func (ms *MutableStateImpl) GetCurrentVersion() int64 { mutable_state_impl.go

1179 > // TODO: can we always return ms.currentVersion here?

1180 > if ms.executionInfo.VersionHistories != nil {

1181 > return ms.currentVersion

1182 > }

1183

1184 if ms.transitionHistoryEnabled && len(ms.executionInfo.TransitionHistory) != 0 {

1234 }

1235

1236 > func (ms *MutableStateImpl) GetCloseVersion() (int64, error) { mutable_state_impl.go

1237 > // TODO: Remove this special handling for zombie workflow.

1238 > // This method should not be called for zombie workflow as it's not considered closed (though it's not running either).

1239 > //

1240 > // However, most callers in this codebase simply check if workflowIsRunning before calling this method, so we cloud reach here

1241 > // even if the workflow is zombie.

1242 > // Most callers are in task executor logic, and we should just prevent any task executor from running when workflow is in zombie state.

1243 > if ms.executionState.State == enumsspb.WORKFLOW_EXECUTION_STATE_ZOMBIE {

1244 return ms.GetLastWriteVersion()

1245 }

1247 // Do NOT use ms.IsWorkflowExecutionRunning() for the check.

1248 // Zombie workflow is not considered running but also not closed.

1249 > if ms.executionState.State != enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED { mutable_state_impl.go

1250 return common.EmptyVersion, serviceerror.NewInternalf("workflow still running, current state: %v", ms.executionState.State.String())

1251 }

1253 // if workflow is closing in the current transation,

1254 // then the last event is closed event and the event version is the close version

1255 > if lastEventVersion, ok := ms.hBuilder.LastEventVersion(); ok { mutable_state_impl.go

1256 > return lastEventVersion, nil mutable_state_impl.go

1257 > }

1258

1259 // We check version history first to prevserve the existing behaior of workflow to minimize risk.

1261 // That assumption is true today, but no necessarily true in the future. We should fix this if we ever

1262 // have such a case.

1263 > if ms.executionInfo.VersionHistories != nil { mutable_state_impl.go

1264 > isEmpty, err := versionhistory.IsCurrentVersionHistoryEmpty(ms.executionInfo.VersionHistories)

1265 > if err != nil {

1266 return common.EmptyVersion, err

1267 }

1268 > if !isEmpty { mutable_state_impl.go

1269 > return ms.GetLastEventVersion() mutable_state_impl.go

1270 > }

1271 }

1272

1287 }

1288

1289 > func (ms *MutableStateImpl) GetLastWriteVersion() (int64, error) { mutable_state_impl.go

1290 > if ms.transitionHistoryEnabled && len(ms.executionInfo.TransitionHistory) != 0 {

1291 > lastVersionedTransition := ms.CurrentVersionedTransition() mutable_state_impl.go

1292 > return lastVersionedTransition.NamespaceFailoverVersion, nil

1293 > }

1294

1295 return ms.GetLastEventVersion()

1296 }

1297

1298 > func (ms *MutableStateImpl) GetLastEventVersion() (int64, error) { mutable_state_impl.go

1299 > if ms.executionInfo.VersionHistories != nil {

1300 > versionHistory, err := versionhistory.GetCurrentVersionHistory(ms.executionInfo.VersionHistories)

1301 > if err != nil {

1302 return 0, err

1303 }

1304 > lastItem, err := versionhistory.GetLastVersionHistoryItem(versionHistory) mutable_state_impl.go

1305 > if err != nil {

1306 return 0, err

1307 }

1308 > return lastItem.GetVersion(), nil mutable_state_impl.go

1309 }

1310

1358 }

1359

1360 > func (ms *MutableStateImpl) GetNamespaceEntry() *namespace.Namespace { mutable_state_impl.go

1361 > return ms.namespaceEntry

1362 > }

1363

1364 // AddHistoryEvent adds any history event to this workflow execution.

1429 }

1430

1431 > func (ms *MutableStateImpl) CurrentTaskQueue() *taskqueuepb.TaskQueue { mutable_state_impl.go

1432 > if ms.IsStickyTaskQueueSet() {

1433 return &taskqueuepb.TaskQueue{

1434 Name: ms.executionInfo.StickyTaskQueue,

1437 }

1438 }

1439 > return &taskqueuepb.TaskQueue{ mutable_state_impl.go

1440 > Name: ms.executionInfo.TaskQueue,

1441 > Kind: enumspb.TASK_QUEUE_KIND_NORMAL,

1442 > }

1443 }

1444

1448 }

1449

1450 > func (ms *MutableStateImpl) ClearStickyTaskQueue() { mutable_state_impl.go

1451 > ms.executionInfo.StickyTaskQueue = ""

1452 > ms.executionInfo.StickyScheduleToStartTimeout = nil

1453 > }

1454

1455 > func (ms *MutableStateImpl) IsStickyTaskQueueSet() bool { mutable_state_impl.go

1456 > return ms.executionInfo.StickyTaskQueue != ""

1457 > }

1458

1459 // TaskQueueScheduleToStartTimeout returns TaskQueue struct and corresponding StartToClose timeout.

1485 }

1486

1487 > func (ms *MutableStateImpl) GetWorkflowType() *commonpb.WorkflowType { mutable_state_impl.go

1488 > wType := &commonpb.WorkflowType{}

1489 > wType.Name = ms.executionInfo.WorkflowTypeName

1490 >

1491 > return wType

1492 > }

1493

1494 func (ms *MutableStateImpl) GetQueryRegistry() historyi.QueryRegistry {

1916 }

1917

1918 > func (ms *MutableStateImpl) Now() time.Time { mutable_state_impl.go

1919 > return ms.timeSource.Now()

1920 > }

1921

1922 // GetWorkflowCloseTime returns workflow closed time, returns a zero time for open workflow

1987 func (ms *MutableStateImpl) GetFirstRunID(

1988 ctx context.Context,

1989 > ) (string, error) { mutable_state_impl.go

1990 > // Prefer the canonical source on ExecutionState; the equivalent field on ExecutionInfo is

1991 > // deprecated. NewMutableStateFromDB backfills ExecutionState from ExecutionInfo on load, so

1992 > // records that pre-date the ExecutionState field also resolve here.

1993 > if firstRunID := ms.executionState.FirstExecutionRunId; firstRunID != "" {

1994 > return firstRunID, nil

1995 > }

1996 if firstRunID := ms.executionInfo.FirstExecutionRunId; firstRunID != "" {

1997 return firstRunID, nil

2071 func (ms *MutableStateImpl) writeEventToCache(

2072 event *historypb.HistoryEvent,

2073 > ) { mutable_state_impl.go

2074 > // For start event: store it here so the recordWorkflowStarted transfer task doesn't need to

2075 > // load it from database.

2076 > // For completion event: store it here so we can communicate the result to parent execution

2077 > // during the processing of DeleteTransferTask without loading this event from database.

2078 > // For Update events: store it here so that Update disposition lookups can be fast.

2079 > ms.eventsCache.PutEvent(

2080 > events.EventKey{

2081 > NamespaceID: namespace.ID(ms.executionInfo.NamespaceId),

2082 > WorkflowID: ms.executionInfo.WorkflowId,

2083 > RunID: ms.executionState.RunId,

2084 > EventID: event.GetEventId(),

2085 > Version: event.GetVersion(),

2086 > },

2087 > event,

2088 > )

2089 > }

2090

2091 > func (ms *MutableStateImpl) HasParentExecution() bool { mutable_state_impl.go

2092 > return ms.executionInfo.ParentNamespaceId != "" && ms.executionInfo.ParentWorkflowId != ""

2093 > }

2094

2095 func (ms *MutableStateImpl) UpdateActivityProgress(

2336

2337 // GetWorkflowTaskByID returns details about the current workflow task by scheduled event ID.

2338 > func (ms *MutableStateImpl) GetWorkflowTaskByID(scheduledEventID int64) *historyi.WorkflowTaskInfo { mutable_state_impl.go

2339 > return ms.workflowTaskManager.GetWorkflowTaskByID(scheduledEventID)

2340 > }

2341

2342 func (ms *MutableStateImpl) GetPendingActivityInfos() map[int64]*persistencespb.ActivityInfo {

2348 }

2349

2350 > func (ms *MutableStateImpl) GetPendingChildExecutionInfos() map[int64]*persistencespb.ChildExecutionInfo { mutable_state_impl.go

2351 > return ms.pendingChildExecutionInfoIDs

2352 > }

2353

2354 func (ms *MutableStateImpl) GetPendingChildIds() map[int64]struct{} {

2360 }

2361

2362 > func (ms *MutableStateImpl) GetPendingRequestCancelExternalInfos() map[int64]*persistencespb.RequestCancelInfo { mutable_state_impl.go

2363 > return ms.pendingRequestCancelInfoIDs

2364 > }

2365

2366 > func (ms *MutableStateImpl) GetPendingSignalExternalInfos() map[int64]*persistencespb.SignalInfo { mutable_state_impl.go

2367 > return ms.pendingSignalInfoIDs

2368 > }

2369

2370 func (ms *MutableStateImpl) GetPendingSignalRequestedIds() []string {

2384 }

2385

2386 > func (ms *MutableStateImpl) HasStartedWorkflowTask() bool { mutable_state_impl.go

2387 > return ms.workflowTaskManager.HasStartedWorkflowTask()

2388 > }

2389

2390 func (ms *MutableStateImpl) GetStartedWorkflowTask() *historyi.WorkflowTaskInfo {

2392 }

2393

2394 > func (ms *MutableStateImpl) IsTransientWorkflowTask() bool { mutable_state_impl.go

2395 > return ms.executionInfo.WorkflowTaskAttempt > 1

2396 > }

2397

2398 func (ms *MutableStateImpl) ClearTransientWorkflowTask() error {

2441 }

2442

2443 > func (ms *MutableStateImpl) GetAssignedBuildId() string { mutable_state_impl.go

2444 > return ms.executionInfo.AssignedBuildId

2445 > }

2446

2447 func (ms *MutableStateImpl) GetInheritedBuildId() string {

2453 }

2454

2455 > func (ms *MutableStateImpl) HasBufferedEvents() bool { mutable_state_impl.go

2456 > return ms.hBuilder.HasBufferEvents()

2457 > }

2458

2459 // HasAnyBufferedEvent returns true if there is at least one buffered event that matches the provided filter.

2469

2470 // GetNextEventID returns next event ID

2471 > func (ms *MutableStateImpl) GetNextEventID() int64 { mutable_state_impl.go

2472 > return ms.hBuilder.NextEventID()

2473 > }

2474

2475 // GetStartedEventIdForLastCompletedWorkflowTask returns last started workflow task event ID

2478 }

2479

2480 > func (ms *MutableStateImpl) IsWorkflowExecutionRunning() bool { mutable_state_impl.go

2481 > switch ms.executionState.State {

2482 > case enumsspb.WORKFLOW_EXECUTION_STATE_CREATED: mutable_state_impl.go

2483 > return true

2484 > case enumsspb.WORKFLOW_EXECUTION_STATE_RUNNING: mutable_state_impl.go

2485 > return true

2486 > case enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED: mutable_state_impl.go

2487 > return false

2488 case enumsspb.WORKFLOW_EXECUTION_STATE_ZOMBIE:

2489 return false

2580 eventType enumspb.EventType,

2581 eventID int64,

2582 > ) { mutable_state_impl.go

2583 > ms.approximateSize -= ms.executionState.Size()

2584 > if ms.executionState.RequestIds == nil {

2585 ms.executionState.RequestIds = make(map[string]*persistencespb.RequestIDInfo, 1)

2586 }

2587 > ms.executionState.RequestIds[requestID] = &persistencespb.RequestIDInfo{ mutable_state_impl.go

2588 > EventType: eventType,

2589 > EventId: eventID,

2590 > }

2591 > if eventType == enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED {

2592 > ms.executionState.CreateRequestId = requestID

2593 > }

2594 > ms.approximateSize += ms.executionState.Size()

2595 }

2596

2887 execution *commonpb.WorkflowExecution,

2888 startRequest *historyservice.StartWorkflowExecutionRequest,

2889 > ) (*historypb.HistoryEvent, error) { mutable_state_impl.go

2890 > return ms.AddWorkflowExecutionStartedEventWithOptions(

2891 > execution,

2892 > startRequest,

2893 > nil, // resetPoints

2894 > "", // prevRunID

2895 > execution.GetRunId(),

2896 > )

2897 > }

2898

2899 func (ms *MutableStateImpl) AddWorkflowExecutionStartedEventWithOptions(

2903 prevRunID string,

2904 firstRunID string,

2905 > ) (*historypb.HistoryEvent, error) { mutable_state_impl.go

2906 > opTag := tag.WorkflowActionWorkflowStarted

2907 > if err := ms.checkMutability(opTag); err != nil {

2908 return nil, err

2909 }

2910

2911 > eventID := ms.GetNextEventID() mutable_state_impl.go

2912 > if eventID != common.FirstEventID {

2913 ms.logger.Warn(mutableStateInvalidHistoryActionMsg, opTag,

2914 tag.WorkflowEventID(eventID),

2917 }

2918

2919 > event := ms.hBuilder.AddWorkflowExecutionStartedEvent( mutable_state_impl.go

2920 > ms.executionState.StartTime.AsTime(),

2921 > startRequest,

2922 > resetPoints,

2923 > prevRunID,

2924 > firstRunID,

2925 > execution.GetRunId(),

2926 > )

2927 > if err := ms.ApplyWorkflowExecutionStartedEvent(

2928 > startRequest.GetParentExecutionInfo().GetClock(),

2929 > execution,

2930 > startRequest.StartRequest.GetRequestId(),

2931 > event,

2932 > ); err != nil {

2933 return nil, err

2934 }

2935

2936 // TODO merge active & passive task generation

2937 > var err error mutable_state_impl.go

2938 > ms.executionInfo.WorkflowExecutionTimerTaskStatus, err = ms.taskGenerator.GenerateWorkflowStartTasks(

2939 > event,

2940 > )

2941 > if err != nil {

2942 return nil, err

2943 }

2944

2945 > if err := ms.taskGenerator.GenerateRecordWorkflowStartedTasks( mutable_state_impl.go

2946 > event,

2947 > ); err != nil {

2948 return nil, err

2949 }

2950

2951 // Versioning Override set on StartWorkflowExecutionRequest

2952 > if startRequest.GetStartRequest().GetVersioningOverride() != nil { mutable_state_impl.go

2953 metrics.WorkerDeploymentVersioningOverrideCounter.With(

2954 ms.metricsHandler.WithTags(

2961 }

2962

2963 > return event, nil mutable_state_impl.go

2964 }

2965

2969 requestID string,

2970 startEvent *historypb.HistoryEvent,

2971 > ) error { mutable_state_impl.go

2972 > if ms.executionInfo.NamespaceId != ms.namespaceEntry.ID().String() {

2973 return serviceerror.NewInternalf("applying conflicting namespace ID: %v != %v",

2974 ms.executionInfo.NamespaceId, ms.namespaceEntry.ID().String())

2975 }

2976 > if ms.executionInfo.WorkflowId != execution.GetWorkflowId() { mutable_state_impl.go

2977 return serviceerror.NewInternalf("applying conflicting workflow ID: %v != %v",

2978 ms.executionInfo.WorkflowId, execution.GetWorkflowId())

2979 }

2980 > if ms.executionState.RunId != execution.GetRunId() { mutable_state_impl.go

2981 return serviceerror.NewInternalf("applying conflicting run ID: %v != %v",

2982 ms.executionState.RunId, execution.GetRunId())

2983 }

2984

2985 > event := startEvent.GetWorkflowExecutionStartedEventAttributes() mutable_state_impl.go

2986 > ms.AttachRequestID(requestID, startEvent.EventType, startEvent.EventId)

2987 >

2988 > ms.approximateSize -= ms.executionInfo.Size()

2989 > ms.executionInfo.FirstExecutionRunId = event.GetFirstExecutionRunId()

2990 > ms.executionInfo.TaskQueue = event.TaskQueue.GetName()

2991 > ms.executionInfo.WorkflowTypeName = event.WorkflowType.GetName()

2992 > ms.executionInfo.WorkflowRunTimeout = event.GetWorkflowRunTimeout()

2993 > ms.executionInfo.WorkflowExecutionTimeout = event.GetWorkflowExecutionTimeout()

2994 > ms.executionInfo.DefaultWorkflowTaskTimeout = event.GetWorkflowTaskTimeout()

2995 > ms.executionInfo.OriginalExecutionRunId = event.GetOriginalExecutionRunId()

2996 >

2997 > ms.approximateSize -= ms.executionState.Size()

2998 > ms.executionState.FirstExecutionRunId = event.GetFirstExecutionRunId()

2999 > if err := ms.addCompletionCallbacks(

3000 > startEvent,

3001 > requestID,

3002 > event.GetCompletionCallbacks(),

3003 > ); err != nil {

3004 return err

3005 }

3006 > if _, err := ms.UpdateWorkflowStateStatus( mutable_state_impl.go

3007 > enumsspb.WORKFLOW_EXECUTION_STATE_CREATED,

3008 > enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING,

3009 > ); err != nil {

3010 return err

3011 }

3012 > ms.executionInfo.LastCompletedWorkflowTaskStartedEventId = common.EmptyEventID mutable_state_impl.go

3013 > ms.executionInfo.LastFirstEventId = startEvent.GetEventId()

3014 >

3015 > ms.executionInfo.WorkflowTaskVersion = common.EmptyVersion

3016 > ms.executionInfo.WorkflowTaskScheduledEventId = common.EmptyEventID

3017 > ms.executionInfo.WorkflowTaskStartedEventId = common.EmptyEventID

3018 > ms.executionInfo.WorkflowTaskRequestId = emptyUUID

3019 > ms.executionInfo.WorkflowTaskTimeout = timestamp.DurationFromSeconds(0)

3020 >

3021 > ms.executionInfo.CronSchedule = event.GetCronSchedule()

3022 >

3023 > if event.ParentWorkflowExecution != nil {

3024 ms.executionInfo.ParentNamespaceId = event.GetParentWorkflowNamespaceId()

3025 ms.executionInfo.ParentWorkflowId = event.ParentWorkflowExecution.GetWorkflowId()

3028 }

3029

3030 > if event.ParentInitiatedEventId != 0 { mutable_state_impl.go

3031 ms.executionInfo.ParentInitiatedId = event.GetParentInitiatedEventId()

3032 > } else { mutable_state_impl.go

3033 > ms.executionInfo.ParentInitiatedId = common.EmptyEventID mutable_state_impl.go

3034 > }

3035

3036 > if event.ParentInitiatedEventVersion != 0 { mutable_state_impl.go

3037 ms.executionInfo.ParentInitiatedVersion = event.GetParentInitiatedEventVersion()

3038 > } else { mutable_state_impl.go

3039 > ms.executionInfo.ParentInitiatedVersion = common.EmptyVersion mutable_state_impl.go

3040 > }

3041

3042 > if event.RootWorkflowExecution != nil { mutable_state_impl.go

3043 ms.executionInfo.RootWorkflowId = event.RootWorkflowExecution.GetWorkflowId()

3044 ms.executionInfo.RootRunId = event.RootWorkflowExecution.GetRunId()

3045 > } else { mutable_state_impl.go

3046 > ms.executionInfo.RootWorkflowId = execution.GetWorkflowId()

3047 > ms.executionInfo.RootRunId = execution.GetRunId()

3048 > }

3049

3050 // todo@time-skipping: apply time skipping to WorkflowStartDelay

3051 > ms.executionInfo.ExecutionTime = timestamppb.New( mutable_state_impl.go

3052 > ms.executionState.StartTime.AsTime().Add(event.GetFirstWorkflowTaskBackoff().AsDuration()),

3053 > )

3054 >

3055 > ms.executionInfo.Attempt = event.GetAttempt()

3056 > if !timestamp.TimeValue(event.GetWorkflowExecutionExpirationTime()).IsZero() {

3057 ms.executionInfo.WorkflowExecutionExpirationTime = event.GetWorkflowExecutionExpirationTime()

3058 }

3059

3060 > var workflowRunTimeoutTime time.Time mutable_state_impl.go

3061 > workflowRunTimeoutDuration := ms.executionInfo.WorkflowRunTimeout.AsDuration()

3062 > // if workflowRunTimeoutDuration == 0 then the workflowRunTimeoutTime will be 0

3063 > // meaning that there is not workflow run timeout

3064 > if workflowRunTimeoutDuration != 0 {

3065 firstWorkflowTaskDelayDuration := event.GetFirstWorkflowTaskBackoff().AsDuration()

3066 workflowRunTimeoutDuration = workflowRunTimeoutDuration + firstWorkflowTaskDelayDuration

3072 }

3073 }

3074 > ms.executionInfo.WorkflowRunExpirationTime = timestamppb.New(workflowRunTimeoutTime) mutable_state_impl.go

3075 >

3076 > if event.RetryPolicy != nil {

3077 ms.executionInfo.HasRetryPolicy = true

3078 ms.executionInfo.RetryBackoffCoefficient = event.RetryPolicy.GetBackoffCoefficient()

3083 }

3084

3085 > ms.executionInfo.AutoResetPoints = rolloverAutoResetPointsWithExpiringTime( mutable_state_impl.go

3086 > event.GetPrevAutoResetPoints(),

3087 > event.GetContinuedExecutionRunId(),

3088 > timestamp.TimeValue(startEvent.GetEventTime()),

3089 > ms.namespaceEntry.Retention(),

3090 > )

3091 >

3092 > if event.Memo != nil {

3093 ms.executionInfo.Memo = event.Memo.GetFields()

3094 }

3095 > if event.SearchAttributes != nil { mutable_state_impl.go

3096 ms.executionInfo.SearchAttributes = event.SearchAttributes.GetIndexedFields()

3097 }

3098

3099 > if event.GetVersioningOverride() != nil { mutable_state_impl.go

3100 if ms.executionInfo.VersioningInfo == nil {

3101 ms.executionInfo.VersioningInfo = &workflowpb.WorkflowExecutionVersioningInfo{}

3136 }

3137

3138 > if event.GetInheritedPinnedVersion() != nil { mutable_state_impl.go

3139 if ms.executionInfo.VersioningInfo == nil {

3140 ms.executionInfo.VersioningInfo = &workflowpb.WorkflowExecutionVersioningInfo{}

3147 // target version upgrade from the started event. This is the same public API

3148 // type, so no conversion needed.

3149 > if event.GetContinuedExecutionRunId() != "" && event.GetInheritedPinnedVersion() != nil { mutable_state_impl.go

3150 ms.executionInfo.DeclinedTargetVersionUpgrade = event.GetDeclinedTargetVersionUpgrade()

3151 }

3152

3153 // Populate the versioningInfo if the inheritedAutoUpgradeInfo is present.

3154 > if event.GetInheritedAutoUpgradeInfo() != nil { mutable_state_impl.go

3155 ms.SetVersioningRevisionNumber(event.GetInheritedAutoUpgradeInfo().GetSourceDeploymentRevisionNumber())

3156 // TODO (Shivam): Remove this once you make SetDeploymentVersion and SetVersioningBehavior methods with nil checks

3164 }

3165

3166 > if inheritedBuildId := event.InheritedBuildId; inheritedBuildId != "" { mutable_state_impl.go

3167 ms.executionInfo.InheritedBuildId = inheritedBuildId

3168 if err := ms.UpdateBuildIdAssignment(inheritedBuildId); err != nil {

3169 return err

3170 }

3171 > } else if event.SourceVersionStamp.GetUseVersioning() && event.SourceVersionStamp.GetBuildId() != "" || mutable_state_impl.go

3172 > ms.GetEffectiveVersioningBehavior() != enumspb.VERSIONING_BEHAVIOR_UNSPECIFIED {

3173 // TODO: [cleanup-old-wv]

3174 limit := ms.config.SearchAttributesSizeOfValueLimit(string(ms.namespaceEntry.Name()))

3181

3182 // This will include override and inheritance, but not transition, because WF never starts with a transition

3183 > ms.executionInfo.WorkerDeploymentName = ms.GetEffectiveDeployment().GetSeriesName() mutable_state_impl.go

3184 >

3185 > if inheritedBuildId := event.InheritedBuildId; inheritedBuildId != "" {

3186 ms.executionInfo.InheritedBuildId = inheritedBuildId

3187 if err := ms.UpdateBuildIdAssignment(inheritedBuildId); err != nil {

3190 }

3191

3192 > ms.executionInfo.MostRecentWorkerVersionStamp = event.SourceVersionStamp mutable_state_impl.go

3193 > ms.executionInfo.Priority = event.Priority

3194 >

3195 > if tsc, stateProp := event.GetTimeSkippingConfig(), event.GetTimeSkippingStatePropagation(); tsc != nil || stateProp.GetInitialSkippedDuration().AsDuration() > 0 {

3196 if err := ms.initTimeSkippingInfo(tsc, stateProp); err != nil {

3197 return err

3199 }

3200

3201 > ms.approximateSize += ms.executionInfo.Size() mutable_state_impl.go

3202 > ms.approximateSize += ms.executionState.Size()

3203 >

3204 > ms.writeEventToCache(startEvent)

3205 > return nil

3206 }

3207

3386 requestID string,

3387 completionCallbacks []*commonpb.Callback,

3388 > ) error { mutable_state_impl.go

3389 > if len(completionCallbacks) == 0 {

3390 > return nil mutable_state_impl.go

3391 > }

3392 if ms.chasmCallbacksEnabled() {

3393 // Initialize chasm tree once for new workflows.

3489 bypassTaskGeneration bool,

3490 workflowTaskType enumsspb.WorkflowTaskType,

3491 > ) (*historyi.WorkflowTaskInfo, error) { mutable_state_impl.go

3492 > opTag := tag.WorkflowActionWorkflowTaskScheduled

3493 > if err := ms.checkMutability(opTag); err != nil {

3494 return nil, err

3495 }

3496 > return ms.workflowTaskManager.AddWorkflowTaskScheduledEvent(bypassTaskGeneration, workflowTaskType) mutable_state_impl.go

3497 }

3498

3538 targetDeploymentVersion *deploymentpb.WorkerDeploymentVersion,

3539 targetRevisionNumber int64,

3540 > ) (*historypb.HistoryEvent, *historyi.WorkflowTaskInfo, error) { mutable_state_impl.go

3541 > opTag := tag.WorkflowActionWorkflowTaskStarted

3542 > if err := ms.checkMutability(opTag); err != nil {

3543 return nil, nil, err

3544 }

3545 > return ms.workflowTaskManager.AddWorkflowTaskStartedEvent(scheduledEventID, requestID, taskQueue, identity, versioningStamp, redirectInfo, skipVersioningCheck, updateReg, targetDeploymentVersion, targetRevisionNumber) mutable_state_impl.go

3546 }

3547

3607 eventID int64,

3608 maxResetPoints int,

3609 > ) bool { mutable_state_impl.go

3610 > resetPoints := ms.executionInfo.AutoResetPoints.GetPoints()

3611 > for _, rp := range resetPoints {

3612 if rp.GetBinaryChecksum() == binaryChecksum && rp.GetBuildId() == buildId {

3613 return false

3617 // todo@time-skipping: time skipping is naturally supported for auto reset points, and need to

3618 // decide if this the best default policy for auto reset points

3619 > newPoint := &workflowpb.ResetPointInfo{ mutable_state_impl.go

3620 > BinaryChecksum: binaryChecksum,

3621 > BuildId: buildId,

3622 > RunId: ms.executionState.GetRunId(),

3623 > FirstWorkflowTaskCompletedId: eventID,

3624 > CreateTime: timestamppb.New(ms.timeSource.Now()),

3625 > Resettable: ms.CheckResettable() == nil,

3626 > }

3627 > ms.executionInfo.AutoResetPoints = &workflowpb.ResetPoints{

3628 > Points: util.SliceTail(append(resetPoints, newPoint), maxResetPoints),

3629 > }

3630 > return true

3631 }

3632

3762 usedVersion *deploymentpb.WorkerDeploymentVersion,

3763 maxSearchAttributeValueSize int,

3764 > ) error { mutable_state_impl.go

3765 > changed, err := ms.addBuildIDAndDeploymentInfoToSearchAttributesWithNoVisibilityTask(stamp, usedVersion, maxSearchAttributeValueSize)

3766 > if err != nil {

3767 return err

3768 }

3769

3770 > if !changed { mutable_state_impl.go

3771 return nil

3772 }

3773 > return ms.taskGenerator.GenerateUpsertVisibilityTask() mutable_state_impl.go

3774 }

3775

3776 > func (ms *MutableStateImpl) loadBuildIds() ([]string, error) { mutable_state_impl.go

3777 > searchAttributes := ms.executionInfo.SearchAttributes

3778 > if searchAttributes == nil {

3779 > return []string{}, nil

3780 > }

3781 saPayload, found := searchAttributes[sadefs.BuildIds]

3782 if !found {

3797 }

3798

3799 > func (ms *MutableStateImpl) loadSearchAttributeString(saName string) (string, error) { mutable_state_impl.go

3800 > searchAttributes := ms.executionInfo.SearchAttributes

3801 > if searchAttributes == nil {

3802 > return "", nil

3803 > }

3804 saPayload, found := searchAttributes[saName]

3805 if !found {

3820 }

3821

3822 > func (ms *MutableStateImpl) loadUsedDeploymentVersions() ([]string, error) { mutable_state_impl.go

3823 > searchAttributes := ms.executionInfo.SearchAttributes

3824 > if searchAttributes == nil {

3825 > return []string{}, nil

3826 > }

3827 saPayload, found := searchAttributes[sadefs.TemporalUsedWorkerDeploymentVersions]

3828 if !found {

3851 existingValues []string,

3852 stamp *commonpb.WorkerVersionStamp,

3853 > ) []string { mutable_state_impl.go

3854 > var newValues []string

3855 > var buildId string

3856 >

3857 > behavior := ms.GetWorkflowVersioningBehaviorSA()

3858 >

3859 > // set up the unversioned or assigned:x sentinels (versioning v2)

3860 > if !stamp.GetUseVersioning() && behavior == enumspb.VERSIONING_BEHAVIOR_UNSPECIFIED { // unversioned workflows may still have non-nil deployment, so we don't check deployment

3861 > newValues = append(newValues, worker_versioning.UnversionedSearchAttribute)

3862 > } else if ms.GetAssignedBuildId() != "" && behavior == enumspb.VERSIONING_BEHAVIOR_UNSPECIFIED {

3863 newValues = append(newValues, worker_versioning.AssignedBuildIdSearchAttribute(ms.GetAssignedBuildId()))

3864 }

3865

3866 // get the most up-to-date pinned entry put it at the front (v3 reachability and v3.1 drainage)

3867 > if behavior == enumspb.VERSIONING_BEHAVIOR_PINNED { mutable_state_impl.go

3868 newValues = append(newValues, worker_versioning.PinnedBuildIdSearchAttribute(ms.GetWorkerDeploymentVersionSA()))

3869 }

3870

3871 // get the build id entry (all versions of versioning)

3872 > if stamp != nil { mutable_state_impl.go

3873 buildId = worker_versioning.VersionStampToBuildIdSearchAttribute(stamp)

3874 }

3875

3876 // add all previous values except for unversioned, assigned, or pinned (there can only be one, and we just added it)

3877 > foundBuildId := false mutable_state_impl.go

3878 > for _, existingValue := range existingValues {

3879 if existingValue == buildId {

3880 foundBuildId = true

3887

3888 // add buildId to the list only if it wasn't there before

3889 > if !foundBuildId && buildId != "" { mutable_state_impl.go

3890 newValues = append(newValues, buildId)

3891 }

3892 > return newValues mutable_state_impl.go

3893 }

3894

3895 > func (ms *MutableStateImpl) addUsedDeploymentVersionToLoadedSearchAttribute(existingValues []string, usedVersion *deploymentpb.WorkerDeploymentVersion) []string { mutable_state_impl.go

3896 > if usedVersion == nil {

3897 > return existingValues

3898 > }

3899

3900 // Get the current deployment version string (already formatted via ExternalWorkerDeploymentVersionToString)

3918 }

3919

3920 > func (ms *MutableStateImpl) saveBuildIds(buildIds []string, maxSearchAttributeValueSize int) error { mutable_state_impl.go

3921 > searchAttributes := ms.executionInfo.SearchAttributes

3922 > if searchAttributes == nil {

3923 > searchAttributes = make(map[string]*commonpb.Payload, 1)

3924 > ms.executionInfo.SearchAttributes = searchAttributes

3925 > }

3926

3927 > hasUnversionedOrAssigned := false mutable_state_impl.go

3928 > if len(buildIds) > 0 { // len is 0 if we are removing the pinned search attribute and the workflow was never unversioned or assigned

3929 > hasUnversionedOrAssigned = worker_versioning.IsUnversionedOrAssignedBuildIdSearchAttribute(buildIds[0])

3930 > }

3931 > for {

3932 > saPayload, err := sadefs.EncodeValue(buildIds, enumspb.INDEXED_VALUE_TYPE_KEYWORD_LIST)

3933 > if err != nil {

3934 return err

3935 }

3936 > if len(buildIds) == 0 || len(saPayload.GetData()) <= maxSearchAttributeValueSize { mutable_state_impl.go

3937 > ms.updateSearchAttributes(map[string]*commonpb.Payload{sadefs.BuildIds: saPayload})

3938 > break

3939 }

3940 if len(buildIds) == 1 {

3947 }

3948 }

3949 > return nil mutable_state_impl.go

3950 }

3951

4025 usedVersion *deploymentpb.WorkerDeploymentVersion,

4026 maxSearchAttributeValueSize int,

4027 > ) (bool, error) { mutable_state_impl.go

4028 > // get all the existing SAs

4029 > existingBuildIds, err := ms.loadBuildIds()

4030 > if err != nil {

4031 return false, err

4032 }

4033 > existingUsedDeploymentVersions, err := ms.loadUsedDeploymentVersions() mutable_state_impl.go

4034 > if err != nil {

4035 return false, err

4036 }

4037 > existingDeployment, err := ms.loadSearchAttributeString(sadefs.TemporalWorkerDeployment) mutable_state_impl.go

4038 > if err != nil {

4039 return false, err

4040 }

4041 > existingVersion, err := ms.loadSearchAttributeString(sadefs.TemporalWorkerDeploymentVersion) mutable_state_impl.go

4042 > if err != nil {

4043 return false, err

4044 }

4045 > existingBehavior, err := ms.loadSearchAttributeString(sadefs.TemporalWorkflowVersioningBehavior) mutable_state_impl.go

4046 > if err != nil {

4047 return false, err

4048 }

4049

4050 // modify them

4051 > modifiedBuildIds := ms.addBuildIdToLoadedSearchAttribute(existingBuildIds, stamp) mutable_state_impl.go

4052 > modifiedUsedDeploymentVersions := ms.addUsedDeploymentVersionToLoadedSearchAttribute(existingUsedDeploymentVersions, usedVersion)

4053 > modifiedDeployment := ms.GetWorkerDeploymentSA()

4054 > modifiedVersion := ms.GetWorkerDeploymentVersionSA()

4055 > modifiedBehavior := ""

4056 > if b := ms.GetWorkflowVersioningBehaviorSA(); b != enumspb.VERSIONING_BEHAVIOR_UNSPECIFIED {

4057 modifiedBehavior = b.String()

4058 }

4059

4060 // check equality

4061 > if slices.Equal(existingBuildIds, modifiedBuildIds) && mutable_state_impl.go

4062 > slices.Equal(existingUsedDeploymentVersions, modifiedUsedDeploymentVersions) &&

4063 > existingDeployment == modifiedDeployment &&

4064 > existingVersion == modifiedVersion &&

4065 > existingBehavior == modifiedBehavior {

4066 return false, nil

4067 }

4068

4069 // save build ids if changed

4070 > if !slices.Equal(existingBuildIds, modifiedBuildIds) { mutable_state_impl.go

4071 > err = ms.saveBuildIds(modifiedBuildIds, maxSearchAttributeValueSize)

4072 > if err != nil {

4073 return false, err // if err != nil, nothing will be written

4074 }

4076

4077 // save used deployment versions if changed

4078 > if !slices.Equal(existingUsedDeploymentVersions, modifiedUsedDeploymentVersions) { mutable_state_impl.go

4079 err = ms.saveUsedDeploymentVersions(modifiedUsedDeploymentVersions, maxSearchAttributeValueSize)

4080 if err != nil {

4084

4085 // save deployment search attributes if changed

4086 > if existingDeployment != modifiedDeployment || mutable_state_impl.go

4087 > existingVersion != modifiedVersion ||

4088 > existingBehavior != modifiedBehavior {

4089 err = ms.saveDeploymentSearchAttributes(modifiedDeployment, modifiedVersion, modifiedBehavior, maxSearchAttributeValueSize)

4090 if err != nil {

4092 }

4093 }

4094 > return true, nil mutable_state_impl.go

4095 }

4096

4098

4099 // CheckResettable check if workflow can be reset

4100 > func (ms *MutableStateImpl) CheckResettable() error { mutable_state_impl.go

4101 > if len(ms.GetPendingChildExecutionInfos()) > 0 {

4102 return serviceerror.NewInvalidArgument("it is not allowed resetting to a point that workflow has pending child workflow.")

4103 }

4104 > if len(ms.GetPendingRequestCancelExternalInfos()) > 0 { mutable_state_impl.go

4105 return serviceerror.NewInvalidArgument("it is not allowed resetting to a point that workflow has pending request cancel.")

4106 }

4107 > if len(ms.GetPendingSignalExternalInfos()) > 0 { mutable_state_impl.go

4108 return serviceerror.NewInvalidArgument("it is not allowed resetting to a point that workflow has pending signals to send.")

4109 }

4110 > return nil mutable_state_impl.go

4111 }

4112

4115 request *workflowservice.RespondWorkflowTaskCompletedRequest,

4116 limits historyi.WorkflowTaskCompletionLimits,

4117 > ) (*historypb.HistoryEvent, error) { mutable_state_impl.go

4118 > opTag := tag.WorkflowActionWorkflowTaskCompleted

4119 > if err := ms.checkMutability(opTag); err != nil {

4120 return nil, err

4121 }

4122 > return ms.workflowTaskManager.AddWorkflowTaskCompletedEvent(workflowTask, request, limits) mutable_state_impl.go

4123 }

4124

4664 // GenerateActivityCancelCommandsForClose generates WorkerCommandsTasks to cancel all

4665 // in-flight activities that have a worker control queue.

4666 > func (ms *MutableStateImpl) GenerateActivityCancelCommandsForClose() error { mutable_state_impl.go

4667 > if !ms.config.EnableCancelActivityWorkerCommand(ms.namespaceEntry.Name().String()) {

4668 > return nil mutable_state_impl.go

4669 > }

4670

4671 // Cancel commands are best-effort and only dispatched on the active cluster.

4829 command *commandpb.CompleteWorkflowExecutionCommandAttributes,

4830 newExecutionRunID string,

4831 > ) (*historypb.HistoryEvent, error) { mutable_state_impl.go

4832 > opTag := tag.WorkflowActionWorkflowCompleted

4833 > if err := ms.checkMutability(opTag); err != nil {

4834 return nil, err

4835 }

4836

4837 > event, batchID := ms.hBuilder.AddCompletedWorkflowEvent(workflowTaskCompletedEventID, command, newExecutionRunID) mutable_state_impl.go

4838 > if err := ms.ApplyWorkflowExecutionCompletedEvent(batchID, event); err != nil {

4839 return nil, err

4840 }

4841 // TODO merge active & passive task generation

4842 > if err := ms.taskGenerator.GenerateWorkflowCloseTasks( mutable_state_impl.go

4843 > event.GetEventTime().AsTime(),

4844 > false,

4845 > false, // skipCloseTransferTask

4846 > ); err != nil {

4847 return nil, err

4848 }

4849 > return event, nil mutable_state_impl.go

4850 }

4851

4853 batchID int64,

4854 event *historypb.HistoryEvent,

4855 > ) error { mutable_state_impl.go

4856 > if _, err := ms.UpdateWorkflowStateStatus(

4857 > enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED,

4858 > enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED,

4859 > ); err != nil {

4860 return err

4861 }

4862 > ms.executionInfo.CompletionEventBatchId = batchID // Used when completion event needs to be loaded from database mutable_state_impl.go

4863 > ms.executionInfo.NewExecutionRunId = event.GetWorkflowExecutionCompletedEventAttributes().GetNewExecutionRunId()

4864 > ms.executionInfo.CloseTime = event.GetEventTime()

4865 > ms.ClearStickyTaskQueue()

4866 > ms.writeEventToCache(event)

4867 > return ms.processCloseCallbacks()

4868 }

4869

6305 newExecutionStartTime time.Time,

6306 namespaceRetention time.Duration,

6307 > ) *workflowpb.ResetPoints { mutable_state_impl.go

6308 > if resetPoints.GetPoints() == nil {

6309 > return resetPoints

6310 > }

6311 newPoints := make([]*workflowpb.ResetPointInfo, 0, len(resetPoints.Points))

6312 // For continue-as-new, new execution start time is the same as previous execution close time,

7043 }

7044

7045 > func (ms *MutableStateImpl) RemoveReportedProblemsSearchAttribute() error { mutable_state_impl.go

7046 > if ms.executionInfo.SearchAttributes == nil {

7047 return nil

7048 }

7049

7050 > temporalReportedProblems := ms.executionInfo.SearchAttributes[sadefs.TemporalReportedProblems] mutable_state_impl.go

7051 > if temporalReportedProblems == nil {

7052 > return nil

7053 > }

7054

7055 // Log the removal of the search attribute

7140 }

7141

7142 > func (ms *MutableStateImpl) AddExternalPayloadSize(size int64) { mutable_state_impl.go

7143 > if ms.executionInfo.ExecutionStats == nil {

7144 ms.executionInfo.ExecutionStats = &persistencespb.ExecutionStats{}

7145 }

7146 > ms.executionInfo.ExecutionStats.ExternalPayloadSize += size mutable_state_impl.go

7147 }

7148

7151 }

7152

7153 > func (ms *MutableStateImpl) AddExternalPayloadCount(count int64) { mutable_state_impl.go

7154 > if ms.executionInfo.ExecutionStats == nil {

7155 ms.executionInfo.ExecutionStats = &persistencespb.ExecutionStats{}

7156 }

7157 > ms.executionInfo.ExecutionStats.ExternalPayloadCount += count mutable_state_impl.go

7158 }

7159

7189 // processCloseCallbacks triggers "WorkflowClosed" callbacks, applying the state machine transition that schedules

7190 // callback tasks.

7191 > func (ms *MutableStateImpl) processCloseCallbacks() error { mutable_state_impl.go

7192 > resetRunID := ms.GetExecutionInfo().GetResetRunId()

7193 > if ms.GetExecutionInfo().GetWorkflowWasReset() && resetRunID != "" && resetRunID != ms.executionState.RunId {

7194 return nil

7195 }

7199 // were enabled can still be triggered even if the EnableCHASMCallbacks dynamic config is later

7200 // turned off. Once created in CHASM, callbacks should always be processed as long as CHASM is enabled.

7201 > if ms.ChasmEnabled() { mutable_state_impl.go

7202 if err := ms.processCloseCallbacksChasm(); err != nil {

7203 return err

7206

7207 // Always process HSM callbacks as well (a workflow can have both)

7208 > return ms.processCloseCallbacksHsm() mutable_state_impl.go

7209 }

7210

7211 // processCloseCallbacksHsm triggers "WorkflowClosed" callbacks using the HSM implementation.

7212 > func (ms *MutableStateImpl) processCloseCallbacksHsm() error { mutable_state_impl.go

7213 > coll := callbacks.MachineCollection(ms.HSM())

7214 > for _, node := range coll.List() {

7215 cb, err := coll.Data(node.Key.ID)

7216 if err != nil {

7228 }

7229 }

7230 > return nil mutable_state_impl.go

7231 }

7232

7254 func (ms *MutableStateImpl) AddTasks(

7255 newTasks ...tasks.Task,

7256 > ) { mutable_state_impl.go

7257 > now := ms.Now()

7258 > for _, task := range newTasks {

7259 > if chasmTask, ok := task.(*tasks.ChasmTask); ok &&

7260 > chasmTask.GetCategory() == tasks.CategoryVisibility &&

7261 > ms.stateInDB == enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED {

7262 softassert.Fail(ms.logger, "CHASM visibility task added on already-closed execution")

7263 }

7264

7265 > category := task.GetCategory() mutable_state_impl.go

7266 > // Drop tasks scheduled too far in the future. VisibilityTime hasn't been

7267 > // shifted to wall-clock yet (the conversion runs below), so both sides are

7268 > // virtual here; the difference is frame-invariant (skip cancels). Keep

7269 > // `now` from ms.Now() so both stay in the same frame.

7270 > if category.Type() == tasks.CategoryTypeScheduled &&

7271 > task.GetVisibilityTime().Sub(now) > maxScheduledTaskDuration {

7272 ms.logger.Info("Dropped long duration scheduled task.", tasks.Tags(task)...)

7273 continue

7279 // vs. real distinction. The CategoryTypeScheduled drop-check above runs first so it

7280 // compares virtual-vs-virtual (now is also virtual).

7281 > if category.Type() == tasks.CategoryTypeScheduled { mutable_state_impl.go

7282 > task.SetVisibilityTime(ms.ToRealTime(task.GetVisibilityTime())) mutable_state_impl.go

7283 > }

7284

7285 > if chasmPureTask, ok := task.(*tasks.ChasmTaskPure); ok { mutable_state_impl.go

7286 ms.chasmPureTasks = append(ms.chasmPureTasks, chasmPureTask)

7287 maxPureTasks := ms.config.ChasmMaxInMemoryPureTasks()

7296 }

7297

7298 > ms.InsertTasks[category] = append(ms.InsertTasks[category], task) mutable_state_impl.go

7299 }

7300 }

7360 }

7361

7362 > func (ms *MutableStateImpl) RemoveSpeculativeWorkflowTaskTimeoutTask() { mutable_state_impl.go

7363 > if ms.speculativeWorkflowTaskTimeoutTask != nil {

7364 // Cancelling task prevents it from being submitted to scheduler in memoryScheduledQueue.

7365 ms.speculativeWorkflowTaskTimeoutTask.Cancel()

7368 }

7369

7370 > func (ms *MutableStateImpl) SetWorkflowTaskScheduleToStartTimeoutTask(task *tasks.WorkflowTaskTimeoutTask) { mutable_state_impl.go

7371 > ms.wftScheduleToStartTimeoutTask = task

7372 > }

7373

7374 > func (ms *MutableStateImpl) SetWorkflowTaskStartToCloseTimeoutTask(task *tasks.WorkflowTaskTimeoutTask) { mutable_state_impl.go

7375 > ms.wftStartToCloseTimeoutTask = task

7376 > }

7377

7378 > func (ms *MutableStateImpl) GetWorkflowTaskScheduleToStartTimeoutTask() *tasks.WorkflowTaskTimeoutTask { mutable_state_impl.go

7379 > return ms.wftScheduleToStartTimeoutTask

7380 > }

7381

7382 > func (ms *MutableStateImpl) GetWorkflowTaskStartToCloseTimeoutTask() *tasks.WorkflowTaskTimeoutTask { mutable_state_impl.go

7383 > return ms.wftStartToCloseTimeoutTask

7384 > }

7385

7386 > func (ms *MutableStateImpl) GetWorkflowStateStatus() (enumsspb.WorkflowExecutionState, enumspb.WorkflowExecutionStatus) { mutable_state_impl.go

7387 > return ms.executionState.State, ms.executionState.Status

7388 > }

7389

7390 func (ms *MutableStateImpl) UpdateWorkflowStateStatus(

7391 state enumsspb.WorkflowExecutionState,

7392 status enumspb.WorkflowExecutionStatus,

7393 > ) (bool, error) { mutable_state_impl.go

7394 > if state == ms.executionState.State && status == ms.executionState.Status {

7395 > return false, nil mutable_state_impl.go

7396 > }

7397 > if state != enumsspb.WORKFLOW_EXECUTION_STATE_ZOMBIE && mutable_state_impl.go

7398 > ms.executionState.State != enumsspb.WORKFLOW_EXECUTION_STATE_ZOMBIE {

7399 > // Suppress and Revive workflows are cluster local operations. mutable_state_impl.go

7400 > ms.executionStateUpdated = true

7401 > ms.visibilityUpdated = true // workflow status & state change triggers visibility change as well

7402 > }

7403 > return true, setStateStatus(ms.executionState, state, status) mutable_state_impl.go

7404 }

7405

7407 // However, certain in-memory changes (e.g. speculative workflow task) won't be cleared before releasing

7408 // the lock and have to be excluded from the check.

7409 > func (ms *MutableStateImpl) IsDirty() bool { mutable_state_impl.go

7410 > return ms.hBuilder.IsDirty() ||

7411 > len(ms.InsertTasks) > 0 ||

7412 > (ms.stateMachineNode != nil && ms.stateMachineNode.Dirty()) ||

7413 > ms.chasmTree.IsDirty()

7414 > }

7415

7416 // isStateDirty is used upon closing transaction to determine if application data has been updated, and

7417 // mutable state should move to a new versioned transition.

7418 > func (ms *MutableStateImpl) isStateDirty() bool { mutable_state_impl.go

7419 > // TODO: we need to track more workflow state changes

7420 > // e.g. changes to executionInfo.CancelRequested

7421 > // They are mostly covered by history builder check today.

7422 > return ms.hBuilder.IsDirty() ||

7423 > len(ms.activityInfosUserDataUpdated) > 0 ||

7424 > len(ms.deleteActivityInfos) > 0 ||

7425 > len(ms.timerInfosUserDataUpdated) > 0 ||

7426 > len(ms.deleteTimerInfos) > 0 ||

7427 > len(ms.updateChildExecutionInfos) > 0 ||

7428 > len(ms.deleteChildExecutionInfos) > 0 ||

7429 > len(ms.updateRequestCancelInfos) > 0 ||

7430 > len(ms.deleteRequestCancelInfos) > 0 ||

7431 > len(ms.updateSignalInfos) > 0 ||

7432 > len(ms.deleteSignalInfos) > 0 ||

7433 > len(ms.updateSignalRequestedIDs) > 0 ||

7434 > len(ms.deleteSignalRequestedIDs) > 0 ||

7435 > len(ms.updateInfoUpdated) > 0 ||

7436 > ms.visibilityUpdated ||

7437 > ms.executionStateUpdated ||

7438 > ms.workflowTaskUpdated ||

7439 > (ms.stateMachineNode != nil && ms.stateMachineNode.Dirty()) ||

7440 > ms.chasmTree.IsStateDirty() ||

7441 > ms.isResetStateUpdated ||

7442 > ms.timeSkippingInfoUpdated

7443 > }

7444

7445 func (ms *MutableStateImpl) IsTransitionHistoryEnabled() bool {

7449 func (ms *MutableStateImpl) StartTransaction(

7450 namespaceEntry *namespace.Namespace,

7451 > ) (bool, error) { mutable_state_impl.go

7452 > if ms.IsDirty() {

7453 ms.logger.Error("MutableState encountered dirty transaction",

7454 tag.WorkflowNamespaceID(ms.executionInfo.NamespaceId),

7461 }

7462

7463 > ms.transitionHistoryEnabled = ms.config.EnableTransitionHistory(namespaceEntry.Name().String()) mutable_state_impl.go

7464 >

7465 > namespaceEntry, err := ms.startTransactionHandleNamespaceMigration(namespaceEntry)

7466 > if err != nil {

7467 return false, err

7468 }

7469 > ms.namespaceEntry = namespaceEntry mutable_state_impl.go

7470 > if err := ms.UpdateCurrentVersion(namespaceEntry.FailoverVersion(ms.executionInfo.WorkflowId), false); err != nil {

7471 return false, err

7472 }

7473

7474 > flushBeforeReady, err := ms.startTransactionHandleWorkflowTaskFailover() mutable_state_impl.go

7475 > if err != nil {

7476 return false, err

7477 }

7478

7479 > return flushBeforeReady, nil mutable_state_impl.go

7480 }

7481

7536 ctx context.Context,

7537 transactionPolicy historyi.TransactionPolicy,

7538 > ) (*persistence.WorkflowSnapshot, []*persistence.WorkflowEvents, error) { mutable_state_impl.go

7539 > result, err := ms.closeTransaction(ctx, transactionPolicy)

7540 > if err != nil {

7541 > return nil, nil, err mutable_state_impl.go

7542 > }

7543

7544 if len(result.bufferEvents) > 0 {

7599 func (ms *MutableStateImpl) updateSearchAttributes(

7600 updatedPayloadMap map[string]*commonpb.Payload,

7601 > ) { mutable_state_impl.go

7602 > ms.executionInfo.SearchAttributes = payload.MergeMapOfPayload(

7603 > ms.executionInfo.SearchAttributes,

7604 > updatedPayloadMap,

7605 > )

7606 > ms.visibilityUpdated = true

7607 > }

7608

7609 func (ms *MutableStateImpl) updateMemo(

7628 func (ms *MutableStateImpl) SetContextMetadata(

7629 ctx context.Context,

7630 > ) { mutable_state_impl.go

7631 > switch ms.chasmTree.ArchetypeID() {

7632 > case chasm.WorkflowArchetypeID, chasm.UnspecifiedArchetypeID: mutable_state_impl.go

7633 > // Set workflow type

7634 > if wfType := ms.GetWorkflowType(); wfType != nil && wfType.GetName() != "" {

7635 > contextutil.ContextMetadataSet(ctx, contextutil.MetadataKeyWorkflowType, wfType.GetName()) mutable_state_impl.go

7636 > }

7637

7638 // Set workflow task queue

7639 > if ms.executionInfo != nil && ms.executionInfo.TaskQueue != "" { mutable_state_impl.go

7640 > contextutil.ContextMetadataSet(ctx, contextutil.MetadataKeyWorkflowTaskQueue, ms.executionInfo.TaskQueue) mutable_state_impl.go

7641 > }

7642

7643 > for _, activityID := range contextutil.ContextMetadataGetMarkedActivityIDs(ctx) { mutable_state_impl.go

7644 if ai, ok := ms.GetActivityByActivityID(activityID); ok {

7645 contextutil.ContextMetadataSet(ctx, contextutil.ActivityTypeKey(ai.ScheduledEventId), ai.ActivityType.GetName())

7660 ctx context.Context,

7661 transactionPolicy historyi.TransactionPolicy,

7662 > ) (closeTransactionResult, error) { mutable_state_impl.go

7663 > ms.SetContextMetadata(ctx)

7664 >

7665 > if err := ms.closeTransactionWithPolicyCheck(

7666 > transactionPolicy,

7667 > ); err != nil {

7668 return closeTransactionResult{}, err

7669 }

7670

7671 > if err := ms.closeTransactionHandleWorkflowTask( mutable_state_impl.go

7672 > transactionPolicy,

7673 > ); err != nil {

7674 return closeTransactionResult{}, err

7675 }

7681 // and need to reconsider the sequence of time skipping close trx handling in this function

7682 // when supporting chasm.

7683 > regenTimerTasksForWorkflowTimeSkipping := ms.closeTransactionHandleWorkflowTimeSkipping(ctx, transactionPolicy) mutable_state_impl.go

7684 >

7685 > // Save if the state is dirty before closeTransactionPrepareEvents since it flushes the buffer

7686 > // events, and therefore change the dirty state.

7687 > isStateDirty := ms.isStateDirty()

7688 >

7689 > // closeTransactionPrepareEvents must be called after closeTransactionHandleWorkflowTask because

7690 > // the latter might fail the workflow task and buffered events must be flushed afterwards.

7691 > // We need to save the value of ms.isStateDirty() before calling closeTransactionPrepareEvents

7692 > // because flushing the buffered events might change the dirty state.

7693 > workflowEventsSeq, eventBatches, bufferEvents, clearBuffer, err := ms.closeTransactionPrepareEvents(transactionPolicy)

7694 > if err != nil {

7695 > return closeTransactionResult{}, err mutable_state_impl.go

7696 > }

7697

7698 // Stamp events with the caller's principal. Only do this on the active

7798 func (ms *MutableStateImpl) closeTransactionHandleWorkflowTask(

7799 transactionPolicy historyi.TransactionPolicy,

7800 > ) error { mutable_state_impl.go

7801 > if err := ms.closeTransactionHandleBufferedEventsLimit(

7802 > transactionPolicy,

7803 > ); err != nil {

7804 return err

7805 }

7806

7807 > if err := ms.closeTransactionHandleWorkflowTaskScheduling( mutable_state_impl.go

7808 > transactionPolicy,

7809 > ); err != nil {

7810 return err

7811 }

7812

7813 > return ms.closeTransactionHandleSpeculativeWorkflowTask(transactionPolicy) mutable_state_impl.go

7814 }

7815

7816 func (ms *MutableStateImpl) closeTransactionHandleWorkflowTaskScheduling(

7817 transactionPolicy historyi.TransactionPolicy,

7818 > ) error { mutable_state_impl.go

7819 > if transactionPolicy == historyi.TransactionPolicyPassive ||

7820 > !ms.IsWorkflowExecutionRunning() {

7821 > return nil mutable_state_impl.go

7822 > }

7823

7824 for _, t := range ms.currentTransactionAddedStateMachineEventTypes {

7845 func (ms *MutableStateImpl) closeTransactionHandleSpeculativeWorkflowTask(

7846 transactionPolicy historyi.TransactionPolicy,

7847 > ) error { mutable_state_impl.go

7848 > if transactionPolicy == historyi.TransactionPolicyPassive ||

7849 > !ms.IsWorkflowExecutionRunning() {

7850 > return nil mutable_state_impl.go

7851 > }

7852

7853 // It is important to convert speculative WT to normal before prepareEventsAndReplicationTasks,

8412 func (ms *MutableStateImpl) closeTransactionPrepareEvents(

8413 transactionPolicy historyi.TransactionPolicy,

8414 > ) ([]*persistence.WorkflowEvents, [][]*historypb.HistoryEvent, []*historypb.HistoryEvent, bool, error) { mutable_state_impl.go

8415 > currentBranchToken, err := ms.GetCurrentBranchToken()

8416 > if err != nil {

8417 return nil, nil, nil, false, err

8418 }

8419

8420 > historyMutation, err := ms.hBuilder.Finish(!ms.HasStartedWorkflowTask()) mutable_state_impl.go

8421 > if err != nil {

8422 return nil, nil, nil, false, err

8423 }

8424

8425 // TODO @wxing1292 need more refactoring to make the logic clean

8426 > ms.bufferEventsInDB = historyMutation.MemBufferBatch mutable_state_impl.go

8427 > newBufferBatch := historyMutation.DBBufferBatch

8428 > clearBuffer := historyMutation.DBClearBuffer

8429 > newEventsBatches := historyMutation.DBEventsBatches

8430 > ms.updatePendingEventIDs(historyMutation.ScheduledIDToStartedID, historyMutation.RequestIDToEventID)

8431 >

8432 > workflowEventsSeq := make([]*persistence.WorkflowEvents, len(newEventsBatches))

8433 > historyNodeTxnIDs, err := ms.shard.GenerateTaskIDs(len(newEventsBatches))

8434 > if err != nil {

8435 return nil, nil, nil, false, err

8436 }

8437 > for index, eventBatch := range newEventsBatches { mutable_state_impl.go

8438 > workflowEventsSeq[index] = &persistence.WorkflowEvents{ mutable_state_impl.go

8439 > NamespaceID: ms.executionInfo.NamespaceId,

8440 > WorkflowID: ms.executionInfo.WorkflowId,

8441 > RunID: ms.executionState.RunId,

8442 > BranchToken: currentBranchToken,

8443 > PrevTxnID: ms.executionInfo.LastFirstEventTxnId,

8444 > TxnID: historyNodeTxnIDs[index],

8445 > Events: eventBatch,

8446 > }

8447 > ms.executionInfo.LastFirstEventId = eventBatch[0].GetEventId()

8448 > ms.executionInfo.LastFirstEventTxnId = historyNodeTxnIDs[index]

8449 >

8450 > // Calculate and add the external payload size and count for this batch

8451 > if ms.config.ExternalPayloadsEnabled(ms.GetNamespaceEntry().Name().String()) {

8452 > externalPayloadSize, externalPayloadCount, err := CalculateExternalPayloadSize(eventBatch, ms.metricsHandler)

8453 > if err != nil {

8454 return nil, nil, nil, false, err

8455 }

8456 > ms.AddExternalPayloadSize(externalPayloadSize) mutable_state_impl.go

8457 > ms.AddExternalPayloadCount(externalPayloadCount)

8458 }

8459 }

8460

8461 > if err := ms.validateNoEventsAfterWorkflowFinish( mutable_state_impl.go

8462 > transactionPolicy,

8463 > workflowEventsSeq,

8464 > ); err != nil {

8465 return nil, nil, nil, false, err

8466 }

8467

8468 > if len(workflowEventsSeq) > 0 { mutable_state_impl.go

8469 > lastEvents := workflowEventsSeq[len(workflowEventsSeq)-1].Events mutable_state_impl.go

8470 > lastEvent := lastEvents[len(lastEvents)-1]

8471 > if err := ms.updateWithLastWriteEvent(

8472 > lastEvent,

8473 > transactionPolicy,

8474 > ); err != nil {

8475 > return nil, nil, nil, false, err mutable_state_impl.go

8476 > }

8477 }

8478

8578 scheduledIDToStartedID map[int64]int64,

8579 requestIDToEventID map[string]int64,

8580 > ) error { mutable_state_impl.go

8581 > for scheduledEventID, startedEventID := range scheduledIDToStartedID {

8582 if activityInfo, ok := ms.GetActivityInfo(scheduledEventID); ok {

8583 activityInfo.StartedEventId = startedEventID

8592 }

8593 }

8594 > if len(requestIDToEventID) > 0 { mutable_state_impl.go

8595 var wf *chasmworkflow.Workflow

8596 var chasmCtx chasm.MutableContext

8615 }

8616 }

8617 > return nil mutable_state_impl.go

8618 }

8619

8621 lastEvent *historypb.HistoryEvent,

8622 transactionPolicy historyi.TransactionPolicy,

8623 > ) error { mutable_state_impl.go

8624 > if transactionPolicy == historyi.TransactionPolicyPassive {

8625 // already handled in mutable state.

8626 return nil

8627 }

8628

8629 > currentVersionHistory, err := versionhistory.GetCurrentVersionHistory(ms.executionInfo.VersionHistories) mutable_state_impl.go

8630 > if err != nil {

8631 return err

8632 }

8633 > if err := versionhistory.AddOrUpdateVersionHistoryItem(currentVersionHistory, versionhistory.NewVersionHistoryItem( mutable_state_impl.go

8634 > lastEvent.GetEventId(), lastEvent.GetVersion(),

8635 > )); err != nil {

8636 > return err mutable_state_impl.go

8637 > }

8638

8639 return nil

8646 transactionPolicy historyi.TransactionPolicy,

8647 workflowEventSeq []*persistence.WorkflowEvents,

8648 > ) error { mutable_state_impl.go

8649 > if transactionPolicy == historyi.TransactionPolicyPassive ||

8650 > len(workflowEventSeq) == 0 {

8651 return nil

8652 }

8653

8654 // only do check if workflow is finished

8655 > if ms.executionState.State != enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED { mutable_state_impl.go

8656 return nil

8657 }

8661 // NOTE: do not apply this check on every batch, since transient

8662 // workflow task && workflow finish will be broken (the first batch)

8663 > eventBatch := workflowEventSeq[len(workflowEventSeq)-1].Events mutable_state_impl.go

8664 > lastEvent := eventBatch[len(eventBatch)-1]

8665 > switch lastEvent.GetEventType() {

8666 case enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_COMPLETED,

8667 enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_FAILED,

8669 enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_TERMINATED,

8670 enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_CONTINUED_AS_NEW,

8671 > enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_CANCELED: mutable_state_impl.go

8672 > return nil

8673

8674 default:

8685 func (ms *MutableStateImpl) startTransactionHandleNamespaceMigration(

8686 namespaceEntry *namespace.Namespace,

8687 > ) (*namespace.Namespace, error) { mutable_state_impl.go

8688 > // NOTE:

8689 > // the main idea here is to guarantee that buffered events & namespace migration works

8690 > // e.g. handle buffered events during version 0 => version > 0 by postponing namespace migration

8691 > // * flush buffered events as if namespace is still local

8692 > // * use updated namespace for actual call

8693 >

8694 > lastWriteVersion, err := ms.GetLastWriteVersion()

8695 > if err != nil {

8696 return nil, err

8697 }

8698

8699 // local namespace -> global namespace && with started workflow task

8700 > if lastWriteVersion == common.EmptyVersion && namespaceEntry.FailoverVersion(ms.executionInfo.WorkflowId) > common.EmptyVersion && ms.HasStartedWorkflowTask() { mutable_state_impl.go

8701 localNamespaceMutation := namespace.WithPretendLocalNamespace(

8702 ms.clusterMetadata.GetCurrentClusterName(),

8704 return namespaceEntry.Clone(localNamespaceMutation), nil

8705 }

8706 > return namespaceEntry, nil mutable_state_impl.go

8707 }

8708

8709 > func (ms *MutableStateImpl) startTransactionHandleWorkflowTaskFailover() (bool, error) { mutable_state_impl.go

8710 > if !ms.IsWorkflowExecutionRunning() {

8711 > return false, nil mutable_state_impl.go

8712 > }

8713

8714 // NOTE:

8805 func (ms *MutableStateImpl) closeTransactionWithPolicyCheck(

8806 transactionPolicy historyi.TransactionPolicy,

8807 > ) error { mutable_state_impl.go

8808 > switch transactionPolicy {

8809 > case historyi.TransactionPolicyActive: mutable_state_impl.go

8810 > // Cannot use ms.namespaceEntry.ActiveClusterName() because currentVersion may be updated during this transaction in

8811 > // passive cluster. For example: if passive cluster sees conflict and decided to terminate this workflow. The

8812 > // currentVersion on mutable state would be updated to point to last write version which is current (passive) cluster.

8813 > activeCluster := ms.clusterMetadata.ClusterNameForFailoverVersion(ms.namespaceEntry.IsGlobalNamespace(), ms.GetCurrentVersion())

8814 > currentCluster := ms.clusterMetadata.GetCurrentClusterName()

8815 >

8816 > if activeCluster != currentCluster {

8817 namespaceID := ms.GetExecutionInfo().NamespaceId

8818 return serviceerror.NewNamespaceNotActive(namespaceID, currentCluster, activeCluster)

8819 }

8820 > return nil mutable_state_impl.go

8821 case historyi.TransactionPolicyPassive:

8822 return nil

8839 func (ms *MutableStateImpl) closeTransactionHandleBufferedEventsLimit(

8840 transactionPolicy historyi.TransactionPolicy,

8841 > ) error { mutable_state_impl.go

8842 > if transactionPolicy == historyi.TransactionPolicyPassive ||

8843 > !ms.IsWorkflowExecutionRunning() {

8844 > return nil mutable_state_impl.go

8845 > }

8846

8847 if ms.BufferSizeAcceptable() {

8968 func (ms *MutableStateImpl) checkMutability(

8969 actionTag tag.ZapTag,

8970 > ) error { mutable_state_impl.go

8971 > if !ms.IsWorkflowExecutionRunning() {

8972 ms.logWarn(

8973 mutableStateInvalidHistoryActionMsg,

8979 return ErrWorkflowFinished

8980 }

8981 > return nil mutable_state_impl.go

8982 }

8983

9073 }

9074

9075 > func (ms *MutableStateImpl) CurrentVersionedTransition() *persistencespb.VersionedTransition { mutable_state_impl.go

9076 > return transitionhistory.LastVersionedTransition(ms.executionInfo.TransitionHistory)

9077 > }

9078

9079 func (ms *MutableStateImpl) ApplyMutation(

9586 }

9587

9588 > func (ms *MutableStateImpl) initVersionedTransitionInDB() { mutable_state_impl.go

9589 > if len(ms.executionInfo.TransitionHistory) != 0 {

9590 > ms.versionedTransitionInDB = ms.CurrentVersionedTransition() mutable_state_impl.go

9591 > }

9592 }

9593

9603 //

9604 // Note: Deployment objects are immutable, never change their fields.

9605 > func (ms *MutableStateImpl) GetEffectiveDeployment() *deploymentpb.Deployment { mutable_state_impl.go

9606 > return GetEffectiveDeployment(ms.GetExecutionInfo().GetVersioningInfo())

9607 > }

9608

9609 > func (ms *MutableStateImpl) GetWorkerDeploymentSA() string { mutable_state_impl.go

9610 > versioningInfo := ms.GetExecutionInfo().GetVersioningInfo()

9611 > if override := versioningInfo.GetVersioningOverride(); override != nil {

9612 if v := worker_versioning.GetOverrideTargetDeploymentVersion(override); v != nil {

9613 return v.GetDeploymentName()

9614 }

9615 }

9616 > if v := versioningInfo.GetDeploymentVersion(); v != nil { mutable_state_impl.go

9617 return v.GetDeploymentName()

9618 }

9619 > return ms.GetExecutionInfo().GetWorkerDeploymentName() mutable_state_impl.go

9620 }

9621

9622 > func (ms *MutableStateImpl) GetWorkerDeploymentVersionSA() string { mutable_state_impl.go

9623 > versioningInfo := ms.GetExecutionInfo().GetVersioningInfo()

9624 > if override := versioningInfo.GetVersioningOverride(); override != nil {

9625 if v := worker_versioning.GetOverrideTargetDeploymentVersion(override); v != nil {

9626 return worker_versioning.ExternalWorkerDeploymentVersionToString(v)

9627 }

9628 }

9629 > if v := versioningInfo.GetDeploymentVersion(); v != nil { mutable_state_impl.go

9630 return worker_versioning.ExternalWorkerDeploymentVersionToString(v)

9631 }

9632 //nolint:staticcheck // SA1019: worker versioning v0.31

9633 > return worker_versioning.ExternalWorkerDeploymentVersionToString(worker_versioning.ExternalWorkerDeploymentVersionFromStringV31(versioningInfo.GetVersion())) mutable_state_impl.go

9634 }

9635

9636 > func (ms *MutableStateImpl) GetWorkflowVersioningBehaviorSA() enumspb.VersioningBehavior { mutable_state_impl.go

9637 > if override := ms.executionInfo.GetVersioningInfo().GetVersioningOverride(); override != nil {

9638 if override.GetAutoUpgrade() {

9639 return enumspb.VERSIONING_BEHAVIOR_AUTO_UPGRADE

9644 return override.GetBehavior()

9645 }

9646 > return ms.executionInfo.GetVersioningInfo().GetBehavior() mutable_state_impl.go

9647 }

9648

9649 > func (ms *MutableStateImpl) GetDeploymentTransition() *workflowpb.DeploymentTransition { mutable_state_impl.go

9650 > vi := ms.GetExecutionInfo().GetVersioningInfo()

9651 > if t := vi.GetVersionTransition(); t != nil {

9652 //nolint:staticcheck // SA1019: worker versioning v0.30

9653 ret := &workflowpb.DeploymentTransition{}

9662 }

9663 //nolint:staticcheck // SA1019: worker versioning v0.30

9664 > return ms.GetExecutionInfo().GetVersioningInfo().GetDeploymentTransition() mutable_state_impl.go

9665 }

9666

9672 // 3. Behavior: this is returned when there is no override (most common case). Behavior is

9673 // set based on the worker-sent deployment in the latest WFT completion.

9674 > func (ms *MutableStateImpl) GetEffectiveVersioningBehavior() enumspb.VersioningBehavior { mutable_state_impl.go

9675 > return GetEffectiveVersioningBehavior(ms.GetExecutionInfo().GetVersioningInfo())

9676 > }

9677

9678 // StartDeploymentTransition starts a transition to the given deployment which must be

9923 }

9924

9925 > func (ms *MutableStateImpl) ToRealTime(virtualTime time.Time) time.Time { mutable_state_impl.go

9926 > if virtualTime.IsZero() {

9927 return virtualTime

9928 }

9929 > return virtualTime.Add(-ms.accumulatedSkippedDuration()) mutable_state_impl.go

9930 }