TestRequestCancel_RespondWorkflowTaskCompleted_NoHeartBeat

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 mutable_state_impl.go

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 mutable_state_impl.go

468 > mutableState.approximateSize += activityInfo.Size()

469 > if (activityInfo.TimerTaskStatus & TimerTaskStatusCreatedHeartbeat) > 0 {

470 // Sets last pending timer heartbeat to year 2000.

471 // This ensures at least one heartbeat task will be processed for the pending activity.

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) mutable_state_impl.go

516 > }

517

518 // FirstExecutionRunId was duplicated from ExecutionInfo onto ExecutionState so the dedup /

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.

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 {

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()

1312 }

1313

1314 > func (ms *MutableStateImpl) IsCurrentWorkflowGuaranteed() bool { mutable_state_impl.go

1315 > // stateInDB is used like a bloom filter:

1316 > //

1317 > // 1. stateInDB being created / running meaning that this workflow must be the current

1318 > // workflow (assuming there is no rebuild of mutable state).

1319 > // 2. stateInDB being completed does not guarantee this workflow being the current workflow

1320 > // 3. stateInDB being zombie guarantees this workflow not being the current workflow

1321 > // 4. stateInDB cannot be void, void is only possible when mutable state is just initialized

1322 >

1323 > switch ms.stateInDB {

1324 case enumsspb.WORKFLOW_EXECUTION_STATE_VOID:

1325 return false

1326 case enumsspb.WORKFLOW_EXECUTION_STATE_CREATED:

1327 return true

1328 > case enumsspb.WORKFLOW_EXECUTION_STATE_RUNNING: mutable_state_impl.go

1329 > return true

1330 case enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED:

1331 return false

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 { mutable_state_impl.go

1495 > return ms.QueryRegistry

1496 > }

1497

1498 // VisitUpdates visits mutable state update entries, ordered by the ID of the history event pointed to by the mutable

1499 // state entry. Thus, for example, updates entries in Admitted state will be visited in the order that their Admitted

1500 // events were added to history.

1501 > func (ms *MutableStateImpl) VisitUpdates(visitor func(updID string, updInfo *persistencespb.UpdateInfo)) { mutable_state_impl.go

1502 > type updateEvent struct {

1503 > updId string

1504 > updInfo *persistencespb.UpdateInfo

1505 > eventId int64

1506 > }

1507 > var updateEvents []updateEvent

1508 > for updID, updInfo := range ms.executionInfo.GetUpdateInfos() {

1509 u := updateEvent{

1510 updId: updID,

1520 updateEvents = append(updateEvents, u)

1521 }

1522 > slices.SortFunc(updateEvents, func(u1, u2 updateEvent) int { return cmp.Compare(u1.eventId, u2.eventId) }) mutable_state_impl.go

1523

1524 > for _, u := range updateEvents { mutable_state_impl.go

1525 visitor(u.updId, u.updInfo)

1526 }

1615 func (ms *MutableStateImpl) GetActivityInfo(

1616 scheduledEventID int64,

1617 > ) (*persistencespb.ActivityInfo, bool) { mutable_state_impl.go

1618 > ai, ok := ms.pendingActivityInfoIDs[scheduledEventID]

1619 > return ai, ok

1620 > }

1621

1622 // GetActivityInfoWithTimerHeartbeat gives details about an activity that is currently in progress.

1636 func (ms *MutableStateImpl) GetActivityByActivityID(

1637 activityID string,

1638 > ) (*persistencespb.ActivityInfo, bool) { mutable_state_impl.go

1639 > eventID, ok := ms.pendingActivityIDToEventID[activityID]

1640 > if !ok {

1641 > return nil, false mutable_state_impl.go

1642 > }

1643 return ms.GetActivityInfo(eventID)

1644 }

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

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 {

2096 ai *persistencespb.ActivityInfo,

2097 request *workflowservice.RecordActivityTaskHeartbeatRequest,

2098 > ) { mutable_state_impl.go

2099 > if prev, existed := ms.pendingActivityInfoIDs[ai.ScheduledEventId]; existed {

2100 > ms.approximateSize -= prev.Size()

2101 > }

2102 > ai.Version = ms.GetCurrentVersion()

2103 > ai.LastHeartbeatDetails = request.Details

2104 > now := ms.timeSource.Now()

2105 > ai.LastHeartbeatUpdateTime = timestamppb.New(now)

2106 > ms.updateActivityInfos[ai.ScheduledEventId] = ai

2107 > ms.activityInfosUserDataUpdated[ai.ScheduledEventId] = struct{}{}

2108 > ms.approximateSize += ai.Size()

2109 > ms.syncActivityTasks[ai.ScheduledEventId] = struct{}{}

2110 >

2111 > payloadSize := request.Details.Size()

2112 > if payloadSize > 0 {

2113 > ms.metricsHandler.Counter(metrics.ActivityPayloadSize.Name()).Record(

2114 > int64(payloadSize),

2115 > metrics.OperationTag(metrics.HistoryRecordActivityTaskHeartbeatScope),

2116 > metrics.NamespaceTag(ms.namespaceEntry.Name().String()))

2117 > }

2118 > ms.metricsHandler.Counter(metrics.ActivityHeartbeatCount.Name()).Record(1,

2119 > metrics.OperationTag(metrics.HistoryRecordActivityTaskHeartbeatScope),

2120 > metrics.NamespaceTag(ms.namespaceEntry.Name().String()),

2121 > metrics.StringTag("has_details", strconv.FormatBool(payloadSize > 0)))

2122 }

2123

2188

2189 // UpdateActivityTaskStatusWithTimerHeartbeat updates an activity's timer task status or/and timer heartbeat

2190 > func (ms *MutableStateImpl) UpdateActivityTaskStatusWithTimerHeartbeat(scheduleEventID int64, timerTaskStatus int32, heartbeatTimeoutVisibility *time.Time) error { mutable_state_impl.go

2191 > ai, ok := ms.pendingActivityInfoIDs[scheduleEventID]

2192 > if !ok {

2193 ms.logError(

2194 fmt.Sprintf("unable to find activity event ID: %v in mutable state", scheduleEventID),

2198 }

2199

2200 > ai.TimerTaskStatus = timerTaskStatus mutable_state_impl.go

2201 > ms.updateActivityInfos[ai.ScheduledEventId] = ai

2202 >

2203 > if heartbeatTimeoutVisibility != nil {

2204 ms.pendingActivityTimerHeartbeats[scheduleEventID] = *heartbeatTimeoutVisibility

2205 }

2206 > return nil mutable_state_impl.go

2207 }

2208

2210 func (ms *MutableStateImpl) DeleteActivity(

2211 scheduledEventID int64,

2212 > ) error { mutable_state_impl.go

2213 > if activityInfo, ok := ms.pendingActivityInfoIDs[scheduledEventID]; ok {

2214 > delete(ms.pendingActivityInfoIDs, scheduledEventID)

2215 > delete(ms.pendingActivityTimerHeartbeats, scheduledEventID)

2216 > ms.approximateSize -= activityInfo.Size() + int64SizeBytes

2217 >

2218 > if _, ok = ms.pendingActivityIDToEventID[activityInfo.ActivityId]; ok {

2219 > delete(ms.pendingActivityIDToEventID, activityInfo.ActivityId)

2220 > } else {

2221 ms.logError(

2222 fmt.Sprintf("unable to find activity ID: %v in mutable state", activityInfo.ActivityId),

2235 }

2236

2237 > delete(ms.updateActivityInfos, scheduledEventID) mutable_state_impl.go

2238 > delete(ms.activityInfosUserDataUpdated, scheduledEventID)

2239 > delete(ms.syncActivityTasks, scheduledEventID)

2240 > ms.deleteActivityInfos[scheduledEventID] = struct{}{}

2241 > return nil

2242 }

2243

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 { mutable_state_impl.go

2343 > return ms.pendingActivityInfoIDs

2344 > }

2345

2346 > func (ms *MutableStateImpl) GetPendingTimerInfos() map[string]*persistencespb.TimerInfo { mutable_state_impl.go

2347 > return ms.pendingTimerInfoIDs

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 {

2376 }

2377

2378 > func (ms *MutableStateImpl) HasPendingWorkflowTask() bool { mutable_state_impl.go

2379 > return ms.workflowTaskManager.HasPendingWorkflowTask()

2380 > }

2381

2382 func (ms *MutableStateImpl) GetPendingWorkflowTask() *historyi.WorkflowTaskInfo {

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 { mutable_state_impl.go

2391 > return ms.workflowTaskManager.GetStartedWorkflowTask()

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 {

2449 }

2450

2451 > func (ms *MutableStateImpl) GetMostRecentWorkerVersionStamp() *commonpb.WorkerVersionStamp { mutable_state_impl.go

2452 > return ms.executionInfo.MostRecentWorkerVersionStamp

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:

2487 return false

2534 // GetApproximatePersistedSize returns approximate size of in-memory objects that will be written to

2535 // persistence + size of buffered events in history builder if they will not be flushed

2536 > func (ms *MutableStateImpl) GetApproximatePersistedSize() int { mutable_state_impl.go

2537 > // include buffered events in the size if they will not be flushed

2538 > if ms.BufferSizeAcceptable() && ms.HasStartedWorkflowTask() {

2539 return ms.approximateSize + ms.hBuilder.SizeInBytesOfBufferedEvents()

2540 }

2541 > return ms.approximateSize mutable_state_impl.go

2542 }

2543

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() mutable_state_impl.go

3066 > workflowRunTimeoutDuration = workflowRunTimeoutDuration + firstWorkflowTaskDelayDuration

3067 > workflowRunTimeoutTime = ms.executionState.StartTime.AsTime().Add(workflowRunTimeoutDuration)

3068 >

3069 > workflowExecutionTimeoutTime := timestamp.TimeValue(ms.executionInfo.WorkflowExecutionExpirationTime)

3070 > if !workflowExecutionTimeoutTime.IsZero() && workflowRunTimeoutTime.After(workflowExecutionTimeoutTime) {

3071 workflowRunTimeoutTime = workflowExecutionTimeoutTime

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

3208 > func (ms *MutableStateImpl) IsWorkflowExecutionStatusPaused() bool { mutable_state_impl.go

3209 > return ms.executionState.GetStatus() == enumspb.WORKFLOW_EXECUTION_STATUS_PAUSED

3210 > }

3211

3212 func (ms *MutableStateImpl) AddWorkflowExecutionPausedEvent(

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 { mutable_state_impl.go

3613 > return false

3614 > }

3615 }

3616

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 mutable_state_impl.go

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] mutable_state_impl.go

3782 > if !found {

3783 return []string{}, nil

3784 }

3785 > decoded, err := sadefs.DecodeValue(saPayload, enumspb.INDEXED_VALUE_TYPE_KEYWORD_LIST, false) mutable_state_impl.go

3786 > if err != nil {

3787 return nil, err

3788 }

3789 > if decoded == nil { mutable_state_impl.go

3790 return []string{}, nil

3791 }

3792 > searchAttributeValues, ok := decoded.([]string) mutable_state_impl.go

3793 > if !ok {

3794 return nil, serviceerror.NewInternal("invalid search attribute value stored for BuildIds")

3795 }

3796 > return searchAttributeValues, nil mutable_state_impl.go

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] mutable_state_impl.go

3805 > if !found {

3806 > return "", nil

3807 > }

3808 decoded, err := sadefs.DecodeValue(saPayload, enumspb.INDEXED_VALUE_TYPE_KEYWORD, false)

3809 if err != nil {

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] mutable_state_impl.go

3828 > if !found {

3829 > return []string{}, nil

3830 > }

3831 decoded, err := sadefs.DecodeValue(saPayload, enumspb.INDEXED_VALUE_TYPE_KEYWORD_LIST, false)

3832 if err != nil {

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 { mutable_state_impl.go

3880 foundBuildId = true

3881 }

3882 > if !worker_versioning.IsUnversionedOrAssignedBuildIdSearchAttribute(existingValue) && mutable_state_impl.go

3883 > !strings.HasPrefix(existingValue, worker_versioning.BuildIdSearchAttributePrefixPinned) {

3884 newValues = append(newValues, existingValue)

3885 }

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 mutable_state_impl.go

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

4189 command *commandpb.ScheduleActivityTaskCommandAttributes,

4190 bypassTaskGeneration bool,

4191 > ) (*historypb.HistoryEvent, *persistencespb.ActivityInfo, error) { mutable_state_impl.go

4192 > opTag := tag.WorkflowActionActivityTaskScheduled

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

4194 return nil, nil, err

4195 }

4196

4197 > _, ok := ms.GetActivityByActivityID(command.GetActivityId()) mutable_state_impl.go

4198 > if ok {

4199 ms.logger.Warn(mutableStateInvalidHistoryActionMsg, opTag,

4200 tag.WorkflowEventID(ms.GetNextEventID()),

4203 }

4204

4205 > event, batchID := ms.hBuilder.AddActivityTaskScheduledEvent(workflowTaskCompletedEventID, command, ms.namespaceEntry.Name()) mutable_state_impl.go

4206 > ai, err := ms.ApplyActivityTaskScheduledEvent(batchID, event)

4207 > // TODO merge active & passive task generation

4208 > if !bypassTaskGeneration {

4209 > if err := ms.taskGenerator.GenerateActivityTasks( mutable_state_impl.go

4210 > event.GetEventId(),

4211 > ); err != nil {

4212 return nil, nil, err

4213 }

4214 }

4215

4216 > return event, ai, err mutable_state_impl.go

4217 }

4218

4220 batchID int64,

4221 event *historypb.HistoryEvent,

4222 > ) (*persistencespb.ActivityInfo, error) { mutable_state_impl.go

4223 >

4224 > attributes := event.GetActivityTaskScheduledEventAttributes()

4225 >

4226 > scheduledEventID := event.GetEventId()

4227 > scheduleToCloseTimeout := attributes.GetScheduleToCloseTimeout()

4228 >

4229 > ai := &persistencespb.ActivityInfo{

4230 > Version: event.GetVersion(),

4231 > ScheduledEventId: scheduledEventID,

4232 > ScheduledEventBatchId: batchID,

4233 > ScheduledTime: event.GetEventTime(),

4234 > FirstScheduledTime: event.GetEventTime(),

4235 > StartedEventId: common.EmptyEventID,

4236 > StartVersion: common.EmptyVersion,

4237 > StartedTime: nil,

4238 > ActivityId: attributes.ActivityId,

4239 > ScheduleToStartTimeout: attributes.GetScheduleToStartTimeout(),

4240 > ScheduleToCloseTimeout: scheduleToCloseTimeout,

4241 > StartToCloseTimeout: attributes.GetStartToCloseTimeout(),

4242 > HeartbeatTimeout: attributes.GetHeartbeatTimeout(),

4243 > CancelRequested: false,

4244 > CancelRequestId: common.EmptyEventID,

4245 > LastHeartbeatUpdateTime: nil,

4246 > TimerTaskStatus: TimerTaskStatusNone,

4247 > TaskQueue: attributes.TaskQueue.GetName(),

4248 > HasRetryPolicy: attributes.RetryPolicy != nil,

4249 > Attempt: 1,

4250 > ActivityType: attributes.GetActivityType(),

4251 > Priority: attributes.Priority,

4252 > }

4253 >

4254 > if attributes.UseWorkflowBuildId {

4255 if ms.GetAssignedBuildId() != "" {

4256 // only set when using new versioning

4264 }

4265

4266 > if ai.HasRetryPolicy { mutable_state_impl.go

4267 ai.RetryInitialInterval = attributes.RetryPolicy.GetInitialInterval()

4268 ai.RetryBackoffCoefficient = attributes.RetryPolicy.GetBackoffCoefficient()

4279 }

4280

4281 > ms.addPendingActivityInfo(ai) mutable_state_impl.go

4282 > ms.writeEventToCache(event)

4283 > return ai, nil

4284 }

4285

4286 > func (ms *MutableStateImpl) addPendingActivityInfo(ai *persistencespb.ActivityInfo) { mutable_state_impl.go

4287 > ms.pendingActivityInfoIDs[ai.ScheduledEventId] = ai

4288 > ms.pendingActivityIDToEventID[ai.ActivityId] = ai.ScheduledEventId

4289 > ms.updateActivityInfos[ai.ScheduledEventId] = ai

4290 > ms.activityInfosUserDataUpdated[ai.ScheduledEventId] = struct{}{}

4291 > ms.approximateSize += ai.Size() + int64SizeBytes

4292 > ms.executionInfo.ActivityCount++

4293 > }

4294

4295 func (ms *MutableStateImpl) addStartedEventForTransientActivity(

4296 scheduledEventID int64,

4297 versioningStamp *commonpb.WorkerVersionStamp,

4298 > ) error { mutable_state_impl.go

4299 > ai, ok := ms.GetActivityInfo(scheduledEventID)

4300 > if !ok || ai.StartedEventId != common.TransientEventID {

4301 > return nil mutable_state_impl.go

4302 > }

4303

4304 if versioningStamp == nil {

4346 workerControlTaskQueue string,

4347 startedClock *clockspb.VectorClock,

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

4349 > opTag := tag.WorkflowActionActivityTaskStarted

4350 > err := ms.checkMutability(opTag)

4351 > if err != nil {

4352 return nil, err

4353 }

4354

4355 > var redirectCounter int64 mutable_state_impl.go

4356 > buildId := worker_versioning.BuildIdIfUsingVersioning(versioningStamp)

4357 > if buildId != "" {

4358 // note that if versioningStamp.BuildId is present we know it's not an old versioning worker because matching

4359 // does not pass build ID for old versioning workers to Record*TaskStart.

4369 }

4370

4371 > if deployment != nil { mutable_state_impl.go

4372 ai.LastWorkerDeploymentVersion = worker_versioning.WorkerDeploymentVersionToStringV31(worker_versioning.DeploymentVersionFromDeployment(deployment))

4373 ai.LastDeploymentVersion = worker_versioning.ExternalWorkerDeploymentVersionFromDeployment(deployment)

4374 }

4375

4376 > ai.WorkerControlTaskQueue = workerControlTaskQueue mutable_state_impl.go

4377 > ai.StartedClock = startedClock

4378 >

4379 > if !ai.HasRetryPolicy {

4380 > event := ms.hBuilder.AddActivityTaskStartedEvent( mutable_state_impl.go

4381 > scheduledEventID,

4382 > ai.Attempt,

4383 > requestID,

4384 > identity,

4385 > ai.RetryLastFailure,

4386 > versioningStamp,

4387 > redirectCounter,

4388 > )

4389 > if err := ms.ApplyActivityTaskStartedEvent(event); err != nil {

4390 return nil, err

4391 }

4392 > return event, nil mutable_state_impl.go

4393 }

4394

4418 func (ms *MutableStateImpl) ApplyActivityTaskStartedEvent(

4419 event *historypb.HistoryEvent,

4420 > ) error { mutable_state_impl.go

4421 > attributes := event.GetActivityTaskStartedEventAttributes()

4422 > scheduledEventID := attributes.GetScheduledEventId()

4423 > ai, ok := ms.GetActivityInfo(scheduledEventID)

4424 > if !ok {

4425 ms.logError(

4426 fmt.Sprintf("unable to find activity event id: %v in mutable state", scheduledEventID),

4430 }

4431

4432 > ms.approximateSize -= ai.Size() mutable_state_impl.go

4433 >

4434 > ai.Version = event.GetVersion()

4435 > ai.StartedEventId = event.GetEventId()

4436 > ai.StartVersion = event.GetVersion()

4437 > ai.RequestId = attributes.GetRequestId()

4438 > ai.StartedTime = event.GetEventTime()

4439 > ms.updateActivityInfos[ai.ScheduledEventId] = ai

4440 > ms.activityInfosUserDataUpdated[ai.ScheduledEventId] = struct{}{}

4441 > ms.approximateSize += ai.Size()

4442 >

4443 > err := ms.applyActivityBuildIdRedirect(ai, worker_versioning.BuildIdIfUsingVersioning(attributes.GetWorkerVersion()), attributes.GetBuildIdRedirectCounter())

4444 > return err

4445 }

4446

4447 > func (ms *MutableStateImpl) applyActivityBuildIdRedirect(activityInfo *persistencespb.ActivityInfo, buildId string, redirectCounter int64) error { mutable_state_impl.go

4448 > if buildId == "" {

4449 > return nil // not versioned

4450 > }

4451

4452 if useWf := activityInfo.GetUseWorkflowBuildIdInfo(); useWf != nil {

4617 scheduledEventID int64,

4618 _ string,

4619 > ) (*historypb.HistoryEvent, *persistencespb.ActivityInfo, error) { mutable_state_impl.go

4620 > opTag := tag.WorkflowActionActivityTaskCancelRequested

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

4622 return nil, nil, err

4623 }

4624

4625 > ai, ok := ms.GetActivityInfo(scheduledEventID) mutable_state_impl.go

4626 > if !ok {

4627 // It is possible both started and completed events are buffered for this activity

4628 if !ms.hBuilder.HasActivityFinishEvent(scheduledEventID) {

4638

4639 // Check for duplicate cancellation

4640 > if ok && ai.CancelRequested { mutable_state_impl.go

4641 ms.logWarn(mutableStateInvalidHistoryActionMsg, opTag,

4642 tag.WorkflowEventID(ms.GetNextEventID()),

4649

4650 // At this point we know this is a valid activity cancellation request

4651 > actCancelReqEvent := ms.hBuilder.AddActivityTaskCancelRequestedEvent(workflowTaskCompletedEventID, scheduledEventID) mutable_state_impl.go

4652 >

4653 > if err := ms.ApplyActivityTaskCancelRequestedEvent(actCancelReqEvent); err != nil {

4654 return nil, nil, err

4655 }

4656

4657 > return actCancelReqEvent, ai, nil mutable_state_impl.go

4658 }

4659

4739 func (ms *MutableStateImpl) ApplyActivityTaskCancelRequestedEvent(

4740 event *historypb.HistoryEvent,

4741 > ) error { mutable_state_impl.go

4742 > attributes := event.GetActivityTaskCancelRequestedEventAttributes()

4743 > scheduledEventID := attributes.GetScheduledEventId()

4744 > ai, ok := ms.GetActivityInfo(scheduledEventID)

4745 > if !ok {

4746 // This will only be called on active cluster if activity info is found in mutable state

4747 // Passive side logic should always have activity info in mutable state if this is called, as the only

4751 }

4752

4753 > ms.approximateSize -= ai.Size() mutable_state_impl.go

4754 >

4755 > ai.Version = event.GetVersion()

4756 >

4757 > // - We have the activity dispatched to worker.

4758 > // - The activity might not be heartbeat'ing, but the activity can still call RecordActivityHeartBeat()

4759 > // to see cancellation while reporting progress of the activity.

4760 > ai.CancelRequested = true

4761 >

4762 > ai.CancelRequestId = event.GetEventId()

4763 > ms.updateActivityInfos[ai.ScheduledEventId] = ai

4764 > ms.activityInfosUserDataUpdated[ai.ScheduledEventId] = struct{}{}

4765 > ms.approximateSize += ai.Size()

4766 > return nil

4767 }

4768

4773 details *commonpb.Payloads,

4774 identity string,

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

4776 > opTag := tag.WorkflowActionActivityTaskCanceled

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

4778 return nil, err

4779 }

4780

4781 > ai, ok := ms.GetActivityInfo(scheduledEventID) mutable_state_impl.go

4782 > if !ok || ai.StartedEventId != startedEventID {

4783 ms.logWarn(mutableStateInvalidHistoryActionMsg, opTag,

4784 tag.WorkflowEventID(ms.GetNextEventID()),

4789

4790 // Verify cancel request as well.

4791 > if !ai.CancelRequested { mutable_state_impl.go

4792 ms.logWarn(mutableStateInvalidHistoryActionMsg, opTag,

4793 tag.WorkflowEventID(ms.GetNextEventID()),

4799 }

4800

4801 > if err := ms.addStartedEventForTransientActivity(scheduledEventID, nil); err != nil { mutable_state_impl.go

4802 return nil, err

4803 }

4804 > event := ms.hBuilder.AddActivityTaskCanceledEvent( mutable_state_impl.go

4805 > scheduledEventID,

4806 > startedEventID,

4807 > latestCancelRequestedEventID,

4808 > details,

4809 > identity,

4810 > )

4811 > if err := ms.ApplyActivityTaskCanceledEvent(event); err != nil {

4812 return nil, err

4813 }

4814

4815 > return event, nil mutable_state_impl.go

4816 }

4817

4818 func (ms *MutableStateImpl) ApplyActivityTaskCanceledEvent(

4819 event *historypb.HistoryEvent,

4820 > ) error { mutable_state_impl.go

4821 > attributes := event.GetActivityTaskCanceledEventAttributes()

4822 > scheduledEventID := attributes.GetScheduledEventId()

4823 >

4824 > return ms.DeleteActivity(scheduledEventID)

4825 > }

4826

4827 func (ms *MutableStateImpl) AddCompletedWorkflowEvent(

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

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

7483 ctx context.Context,

7484 transactionPolicy historyi.TransactionPolicy,

7485 > ) (*persistence.WorkflowMutation, []*persistence.WorkflowEvents, error) { mutable_state_impl.go

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

7487 > if err != nil {

7488 return nil, nil, err

7489 }

7490

7491 > if result.skipPersistence { mutable_state_impl.go

7492 if err := ms.cleanupTransaction(); err != nil {

7493 return nil, nil, err

7496 }

7497

7498 > workflowMutation := &persistence.WorkflowMutation{ mutable_state_impl.go

7499 > ExecutionInfo: ms.executionInfo,

7500 > ExecutionState: ms.executionState,

7501 > NextEventID: ms.hBuilder.NextEventID(),

7502 >

7503 > UpsertActivityInfos: ms.updateActivityInfos,

7504 > DeleteActivityInfos: ms.deleteActivityInfos,

7505 > UpsertTimerInfos: ms.updateTimerInfos,

7506 > DeleteTimerInfos: ms.deleteTimerInfos,

7507 > UpsertChildExecutionInfos: ms.updateChildExecutionInfos,

7508 > DeleteChildExecutionInfos: ms.deleteChildExecutionInfos,

7509 > UpsertRequestCancelInfos: ms.updateRequestCancelInfos,

7510 > DeleteRequestCancelInfos: ms.deleteRequestCancelInfos,

7511 > UpsertSignalInfos: ms.updateSignalInfos,

7512 > DeleteSignalInfos: ms.deleteSignalInfos,

7513 > UpsertSignalRequestedIDs: ms.updateSignalRequestedIDs,

7514 > DeleteSignalRequestedIDs: ms.deleteSignalRequestedIDs,

7515 > UpsertChasmNodes: result.chasmNodesMutation.UpdatedNodes,

7516 > DeleteChasmNodes: result.chasmNodesMutation.DeletedNodes,

7517 > NewBufferedEvents: result.bufferEvents,

7518 > ClearBufferedEvents: result.clearBuffer,

7519 >

7520 > Tasks: ms.InsertTasks,

7521 > BestEffortDeleteTasks: ms.BestEffortDeleteTasks,

7522 >

7523 > Condition: ms.nextEventIDInDB,

7524 > DBRecordVersion: ms.dbRecordVersion,

7525 > Checksum: result.checksum,

7526 > }

7527 >

7528 > ms.checksum = result.checksum

7529 > if err := ms.cleanupTransaction(); err != nil {

7530 return nil, nil, err

7531 }

7532 > return workflowMutation, result.workflowEventsSeq, nil mutable_state_impl.go

7533 }

7534

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

7542 }

7543

7544 > if len(result.bufferEvents) > 0 { mutable_state_impl.go

7545 // TODO do we need the functionality to generate snapshot with buffered events?

7546 return nil, nil, softassert.UnexpectedInternalErr(

7551 }

7552

7553 > workflowSnapshot := &persistence.WorkflowSnapshot{ mutable_state_impl.go

7554 > ExecutionInfo: ms.executionInfo,

7555 > ExecutionState: ms.executionState,

7556 > NextEventID: ms.hBuilder.NextEventID(),

7557 >

7558 > ActivityInfos: ms.pendingActivityInfoIDs,

7559 > TimerInfos: ms.pendingTimerInfoIDs,

7560 > ChildExecutionInfos: ms.pendingChildExecutionInfoIDs,

7561 > RequestCancelInfos: ms.pendingRequestCancelInfoIDs,

7562 > SignalInfos: ms.pendingSignalInfoIDs,

7563 > SignalRequestedIDs: ms.pendingSignalRequestedIDs,

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

7565 >

7566 > Tasks: ms.InsertTasks,

7567 >

7568 > Condition: ms.nextEventIDInDB,

7569 > DBRecordVersion: ms.dbRecordVersion,

7570 > Checksum: result.checksum,

7571 > }

7572 >

7573 > ms.checksum = result.checksum

7574 > if err := ms.cleanupTransaction(); err != nil {

7575 return nil, nil, err

7576 }

7577 > return workflowSnapshot, result.workflowEventsSeq, nil mutable_state_impl.go

7578 }

7579

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

7696 }

7699 // cluster — standby (passive) replays events that were already stamped by

7700 // the active side, and we must not overwrite those principals.

7701 > if transactionPolicy == historyi.TransactionPolicyActive { mutable_state_impl.go

7702 > principal := headers.GetPrincipal(ctx) mutable_state_impl.go

7703 > for _, we := range workflowEventsSeq {

7704 > for _, event := range we.Events { mutable_state_impl.go

7705 > // Skip events that already have a principal. Those are previously

7706 > // buffered events (e.g., signals) that were stamped when originally

7707 > // created and are now being flushed into history by a different caller

7708 > // (e.g., the worker completing a workflow task).

7709 > if event.Principal == nil {

7710 > event.Principal = principal

7711 > }

7712 }

7713 }

7714 > for _, event := range bufferEvents { mutable_state_impl.go

7715 event.Principal = principal

7716 }

7719 // CloseTransaction() on chasmTree may update execution state & status,

7720 // so must be called before closeTransactionUpdateTransitionHistory().

7721 > chasmNodesMutation, err := ms.chasmTree.CloseTransaction() mutable_state_impl.go

7722 > if err != nil {

7723 return closeTransactionResult{}, err

7724 }

7725

7726 > if ms.closeTransactionShouldSkipPersistence(isStateDirty, chasmNodesMutation) { mutable_state_impl.go

7727 return closeTransactionResult{

7728 skipPersistence: true,

7730 }

7731

7732 > for nodePath := range chasmNodesMutation.DeletedNodes { mutable_state_impl.go

7733 ms.approximateSize -= ms.chasmNodeSizes[nodePath]

7734 delete(ms.chasmNodeSizes, nodePath)

7735 }

7736 > for nodePath, node := range chasmNodesMutation.UpdatedNodes { mutable_state_impl.go

7737 newSize := len(nodePath) + node.Size()

7738 ms.approximateSize += newSize - ms.chasmNodeSizes[nodePath]

7740 }

7741

7742 > if isStateDirty { mutable_state_impl.go

7743 > if err := ms.closeTransactionUpdateTransitionHistory( mutable_state_impl.go

7744 > transactionPolicy,

7745 > ); err != nil {

7746 return closeTransactionResult{}, err

7747 }

7748 > ms.closeTransactionHandleUnknownVersionedTransition() mutable_state_impl.go

7749 > ms.closeTransactionUpdateLastRunningClock(transactionPolicy, workflowEventsSeq)

7750 }

7751

7752 // todo@TimeSkipping, we can move update versioned transition to inside closeTransactionHandleWorkflowTimeSkipping

7753 > ms.closeTransactionTrackLastUpdateVersionedTransition( mutable_state_impl.go

7754 > transactionPolicy,

7755 > )

7756 >

7757 > ms.closeTransactionTrackTombstones(transactionPolicy, chasmNodesMutation)

7758 >

7759 > // generate tasks

7760 > if err := ms.closeTransactionPrepareTasks(

7761 > transactionPolicy,

7762 > eventBatches,

7763 > clearBuffer,

7764 > regenTimerTasksForWorkflowTimeSkipping,

7765 > ); err != nil {

7766 return closeTransactionResult{}, err

7767 }

7768

7769 > ms.executionInfo.StateTransitionCount += 1 mutable_state_impl.go

7770 > ms.executionInfo.LastUpdateTime = timestamppb.New(ms.timeSource.Now())

7771 >

7772 > // We generate checksum here based on the assumption that the returned

7773 > // snapshot object is considered immutable. As of this writing, the only

7774 > // code that modifies the returned object lives inside Context.resetWorkflowExecution.

7775 > // Currently, the updates done inside Context.resetWorkflowExecution don't

7776 > // impact the checksum calculation.

7777 > checksum := ms.generateChecksum()

7778 >

7779 > if ms.dbRecordVersion == 0 {

7780 > // noop, existing behavior mutable_state_impl.go

7781 > } else { mutable_state_impl.go

7782 > ms.dbRecordVersion += 1 mutable_state_impl.go

7783 > }

7784

7785 > return closeTransactionResult{ mutable_state_impl.go

7786 > workflowEventsSeq: workflowEventsSeq,

7787 > bufferEvents: bufferEvents,

7788 > clearBuffer: clearBuffer,

7789 > checksum: checksum,

7790 > chasmNodesMutation: chasmNodesMutation,

7791 > }, nil

7792 }

7793

7794 > func (ms *MutableStateImpl) closeTransactionShouldSkipPersistence(isStateDirty bool, chasmNodesMutation chasm.NodesMutation) bool { mutable_state_impl.go

7795 > return !ms.IsWorkflow() && !isStateDirty && chasmNodesMutation.IsEmpty()

7796 > }

7797

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

7822 }

7823

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

7825 def, ok := ms.shard.StateMachineRegistry().EventDefinition(t)

7826 if !ok {

7840 }

7841

7842 > return nil mutable_state_impl.go

7843 }

7844

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

7851 }

7854 // because prepareEventsAndReplicationTasks will move internal buffered events to the history,

7855 // and WT related events (WTScheduled, in particular) need to go first.

7856 > return ms.workflowTaskManager.convertSpeculativeWorkflowTaskToNormal() mutable_state_impl.go

7857 }

7858

7859 func (ms *MutableStateImpl) closeTransactionUpdateTransitionHistory(

7860 transactionPolicy historyi.TransactionPolicy,

7861 > ) error { mutable_state_impl.go

7862 > if transactionPolicy != historyi.TransactionPolicyActive {

7863 // TODO: replication/standby logic will need a different way for updating transition history

7864 // when not syncing mutable state

7866 }

7867

7868 > if !ms.transitionHistoryEnabled { mutable_state_impl.go

7869 return nil

7870 }

7871

7872 // handle disable then re-enable of transition history

7873 > if len(ms.executionInfo.TransitionHistory) == 0 && len(ms.executionInfo.PreviousTransitionHistory) != 0 { mutable_state_impl.go

7874 ms.executionInfo.TransitionHistory = ms.executionInfo.PreviousTransitionHistory

7875 ms.executionInfo.PreviousTransitionHistory = nil

7876 }

7877

7878 > ms.executionInfo.TransitionHistory = UpdatedTransitionHistory( mutable_state_impl.go

7879 > ms.executionInfo.TransitionHistory,

7880 > ms.GetCurrentVersion(),

7881 > )

7882 >

7883 > return nil

7884 }

7885

7886 func (ms *MutableStateImpl) closeTransactionTrackLastUpdateVersionedTransition(

7887 transactionPolicy historyi.TransactionPolicy,

7888 > ) { mutable_state_impl.go

7889 > if transactionPolicy != historyi.TransactionPolicyActive {

7890 // TODO: replication/standby logic will need a different way for updating LastUpdatedVersionedTransition

7891 // when reapplying history, especially when history replication tasks got batched.

7893 }

7894

7895 > if !ms.transitionHistoryEnabled { mutable_state_impl.go

7896 // transition history is not enabled

7897 return

7898 }

7899 // transaction closed without any state change

7900 > if len(ms.executionInfo.TransitionHistory) == 0 { mutable_state_impl.go

7901 return

7902 }

7903

7904 > currentVersionedTransition := ms.CurrentVersionedTransition() mutable_state_impl.go

7905 > for activityId := range ms.activityInfosUserDataUpdated {

7906 > ms.updateActivityInfos[activityId].LastUpdateVersionedTransition = currentVersionedTransition mutable_state_impl.go

7907 > }

7908 > for timerId := range ms.timerInfosUserDataUpdated { mutable_state_impl.go

7909 ms.updateTimerInfos[timerId].LastUpdateVersionedTransition = currentVersionedTransition

7910 }

7911 > for _, childInfo := range ms.updateChildExecutionInfos { mutable_state_impl.go

7912 childInfo.LastUpdateVersionedTransition = currentVersionedTransition

7913 }

7914 > for _, cancelInfo := range ms.updateRequestCancelInfos { mutable_state_impl.go

7915 cancelInfo.LastUpdateVersionedTransition = currentVersionedTransition

7916 }

7917 > for _, signalInfo := range ms.updateSignalInfos { mutable_state_impl.go

7918 signalInfo.LastUpdateVersionedTransition = currentVersionedTransition

7919 }

7922 // signal requestedID.

7923 // Deletion of signalRequestID is not replicated today, so we can even drop the check on deleteSignalRequestedIDs

7924 > if len(ms.updateSignalRequestedIDs) != 0 || len(ms.deleteSignalRequestedIDs) != 0 { mutable_state_impl.go

7925 ms.executionInfo.SignalRequestIdsLastUpdateVersionedTransition = currentVersionedTransition

7926 }

7927

7928 > for updateID := range ms.updateInfoUpdated { mutable_state_impl.go

7929 ms.executionInfo.UpdateInfos[updateID].LastUpdateVersionedTransition = currentVersionedTransition

7930 }

7931

7932 > if ms.workflowTaskUpdated { mutable_state_impl.go

7933 > ms.executionInfo.WorkflowTaskLastUpdateVersionedTransition = currentVersionedTransition mutable_state_impl.go

7934 > }

7935

7936 > if ms.visibilityUpdated { mutable_state_impl.go

7937 > ms.executionInfo.VisibilityLastUpdateVersionedTransition = currentVersionedTransition mutable_state_impl.go

7938 > }

7939

7940 > if ms.executionStateUpdated { mutable_state_impl.go

7941 > ms.executionState.LastUpdateVersionedTransition = currentVersionedTransition mutable_state_impl.go

7942 > }

7943

7944 > if ms.timeSkippingInfoUpdated && ms.executionInfo.TimeSkippingInfo != nil { mutable_state_impl.go

7945 ms.executionInfo.TimeSkippingInfo.LastUpdateVersionedTransition = currentVersionedTransition

7946 }

7950 }

7951

7952 > func (ms *MutableStateImpl) closeTransactionHandleUnknownVersionedTransition() { mutable_state_impl.go

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

7954 > if transitionhistory.Compare( mutable_state_impl.go

7955 > ms.versionedTransitionInDB,

7956 > ms.CurrentVersionedTransition(),

7957 > ) != 0 {

7958 > // versioned transition updated in the transaction

7959 > return

7960 > }

7961 }

7962

8017 transactionPolicy historyi.TransactionPolicy,

8018 workflowEventsSeq []*persistence.WorkflowEvents,

8019 > ) { mutable_state_impl.go

8020 > if transactionPolicy != historyi.TransactionPolicyActive {

8021 return

8022 }

8024 // Events can only be generated while mutable state is running,

8025 // so we can update LastRunningClock blindly.

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

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

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

8029 > ms.executionInfo.LastRunningClock = lastEvent.GetTaskId()

8030 > return

8031 > }

8032

8033 > if !ms.IsWorkflowExecutionRunning() && !ms.IsCurrentWorkflowGuaranteed() { mutable_state_impl.go

8034 // If workflow currently is not running and also not running at the beginning of the transaction,

8035 // then don't update the lastRunningClock

8038 }

8039

8040 > ms.executionInfo.LastRunningClock = ms.shard.CurrentVectorClock().GetClock() mutable_state_impl.go

8041 }

8042

8044 transactionPolicy historyi.TransactionPolicy,

8045 chasmNodesMutation chasm.NodesMutation,

8046 > ) { mutable_state_impl.go

8047 > if transactionPolicy != historyi.TransactionPolicyActive {

8048 // Passive/Replication logic will update tombstone list when applying mutable state

8049 // snapshot or mutation.

8051 }

8052

8053 > if !ms.transitionHistoryEnabled { mutable_state_impl.go

8054 // transition history is not enabled

8055 return

8056 }

8057

8058 > if len(ms.executionInfo.TransitionHistory) == 0 { mutable_state_impl.go

8059 // in an unknown state

8060 return

8061 }

8062

8063 > var tombstones []*persistencespb.StateMachineTombstone mutable_state_impl.go

8064 > if ms.stateMachineNode != nil {

8065 > opLog, err := ms.stateMachineNode.OpLog()

8066 > if err != nil {

8067 panic(fmt.Sprintf("Failed to get HSM operation log: %v", err))

8068 }

8069

8070 > for _, op := range opLog { mutable_state_impl.go

8071 if deleteOp, ok := op.(hsm.DeleteOperation); ok {

8072 path := deleteOp.Path()

8095 }

8096

8097 > for scheduledEventID := range ms.deleteActivityInfos { mutable_state_impl.go

8098 > tombstones = append(tombstones, &persistencespb.StateMachineTombstone{ mutable_state_impl.go

8099 > StateMachineKey: &persistencespb.StateMachineTombstone_ActivityScheduledEventId{

8100 > ActivityScheduledEventId: scheduledEventID,

8101 > },

8102 > })

8103 > }

8104 > for timerID := range ms.deleteTimerInfos { mutable_state_impl.go

8105 tombstones = append(tombstones, &persistencespb.StateMachineTombstone{

8106 StateMachineKey: &persistencespb.StateMachineTombstone_TimerId{

8109 })

8110 }

8111 > for initiatedEventId := range ms.deleteChildExecutionInfos { mutable_state_impl.go

8112 tombstones = append(tombstones, &persistencespb.StateMachineTombstone{

8113 StateMachineKey: &persistencespb.StateMachineTombstone_ChildExecutionInitiatedEventId{

8116 })

8117 }

8118 > for initiatedEventId := range ms.deleteRequestCancelInfos { mutable_state_impl.go

8119 tombstones = append(tombstones, &persistencespb.StateMachineTombstone{

8120 StateMachineKey: &persistencespb.StateMachineTombstone_RequestCancelInitiatedEventId{

8123 })

8124 }

8125 > for initiatedEventId := range ms.deleteSignalInfos { mutable_state_impl.go

8126 tombstones = append(tombstones, &persistencespb.StateMachineTombstone{

8127 StateMachineKey: &persistencespb.StateMachineTombstone_SignalExternalInitiatedEventId{

8130 })

8131 }

8132 > for chasmNodePath := range chasmNodesMutation.DeletedNodes { mutable_state_impl.go

8133 tombstones = append(tombstones, &persistencespb.StateMachineTombstone{

8134 StateMachineKey: &persistencespb.StateMachineTombstone_ChasmNodePath{

8142 // which is not supported by today's DB schema.

8143 // TODO: we don't delete updateInfo today. Track them here when we do.

8144 > currentVersionedTransition := ms.CurrentVersionedTransition() mutable_state_impl.go

8145 >

8146 > tombstoneBatch := &persistencespb.StateMachineTombstoneBatch{

8147 > VersionedTransition: currentVersionedTransition,

8148 > StateMachineTombstones: tombstones,

8149 > }

8150 > // As an optimization, we only track the first empty tombstone batch. So we can know the start point of the tombstone batch

8151 > if len(tombstones) > 0 || len(ms.executionInfo.SubStateMachineTombstoneBatches) == 0 {

8152 > ms.executionInfo.SubStateMachineTombstoneBatches = append(ms.executionInfo.SubStateMachineTombstoneBatches, tombstoneBatch)

8153 > }

8154

8155 > ms.totalTombstones += len(tombstones) mutable_state_impl.go

8156 > ms.capTombstoneCount()

8157 }

8158

8159 // capTombstoneCount limits the total number of tombstones stored in the mutable state.

8160 // This method should be called whenever tombstone batch list is updated or synced.

8161 > func (ms *MutableStateImpl) capTombstoneCount() { mutable_state_impl.go

8162 > tombstoneCountLimit := ms.config.MutableStateTombstoneCountLimit()

8163 > for ms.totalTombstones > tombstoneCountLimit &&

8164 > len(ms.executionInfo.SubStateMachineTombstoneBatches) > 0 {

8165 ms.totalTombstones -= len(ms.executionInfo.SubStateMachineTombstoneBatches[0].StateMachineTombstones)

8166 ms.executionInfo.SubStateMachineTombstoneBatches = ms.executionInfo.SubStateMachineTombstoneBatches[1:]

8173 clearBufferEvents bool,

8174 regenerateTimerTasksForTimeSkipping bool,

8175 > ) error { mutable_state_impl.go

8176 > if err := ms.closeTransactionHandleWorkflowResetTask(

8177 > transactionPolicy,

8178 > ); err != nil {

8179 return err

8180 }

8181

8182 > if err := ms.taskGenerator.GenerateDirtySubStateMachineTasks(ms.shard.StateMachineRegistry()); err != nil { mutable_state_impl.go

8183 return err

8184 }

8185

8186 > ms.closeTransactionCollapseVisibilityTasks() mutable_state_impl.go

8187 >

8188 > if err := ms.closeTransactionGenerateChasmRetentionTask(transactionPolicy); err != nil {

8189 return err

8190 }

8195 // regardless of how many activity & user timer created

8196 // so the calculation must be at the very end

8197 > if err := ms.closeTransactionHandleActivityUserTimerTasks(transactionPolicy); err != nil { mutable_state_impl.go

8198 return err

8199 }

8200 > if regenerateTimerTasksForTimeSkipping { mutable_state_impl.go

8201 if err := ms.closeTransactionRegenTimerTasksForWorkflowTimeSkipping(transactionPolicy); err != nil {

8202 return err

8204 }

8205

8206 > return ms.closeTransactionPrepareReplicationTasks(transactionPolicy, eventBatches, clearBufferEvents) mutable_state_impl.go

8207 }

8208

8209 func (ms *MutableStateImpl) closeTransactionGenerateChasmRetentionTask(

8210 transactionPolicy historyi.TransactionPolicy,

8211 > ) error { mutable_state_impl.go

8212 >

8213 > if ms.IsWorkflow() ||

8214 > ms.executionState.State != enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED ||

8215 > ms.stateInDB == enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED {

8216 > return nil mutable_state_impl.go

8217 > }

8218

8219 // Generate retention timer for chasm executions if it's currentely completed

8234 eventBatches [][]*historypb.HistoryEvent,

8235 clearBufferEvents bool,

8236 > ) error { mutable_state_impl.go

8237 > var replicationTasks []tasks.Task

8238 > if ms.config.ReplicationMultipleBatches() {

8239 task, err := ms.eventsToReplicationTask(transactionPolicy, eventBatches)

8240 if err != nil {

8242 }

8243 replicationTasks = append(replicationTasks, task...)

8244 > } else { mutable_state_impl.go

8245 > for _, historyEvents := range eventBatches {

8246 > task, err := ms.eventsToReplicationTask(transactionPolicy, [][]*historypb.HistoryEvent{historyEvents}) mutable_state_impl.go

8247 > if err != nil {

8248 return err

8249 }

8250 > replicationTasks = append(replicationTasks, task...) mutable_state_impl.go

8251 }

8252 }

8253 > replicationTasks = append(replicationTasks, ms.syncActivityToReplicationTask(transactionPolicy)...) mutable_state_impl.go

8254 > replicationTasks = append(replicationTasks, ms.dirtyHSMToReplicationTask(transactionPolicy, eventBatches, clearBufferEvents)...)

8255 >

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

8257 > isWorkflow := archetypeID == chasm.WorkflowArchetypeID

8258 > if !isWorkflow && len(replicationTasks) != 0 {

8259 return softassert.UnexpectedInternalErr(ms.logger, "chasm execution generated workflow replication tasks", nil)

8260 }

8261

8262 > if ms.transitionHistoryEnabled { mutable_state_impl.go

8263 > switch transactionPolicy { mutable_state_impl.go

8264 > case historyi.TransactionPolicyActive: mutable_state_impl.go

8265 > if ms.generateReplicationTask() {

8266 now := time.Now().UTC()

8267 workflowKey := definition.NewWorkflowKey(

8336 }

8337

8338 > if transactionPolicy == historyi.TransactionPolicyPassive && mutable_state_impl.go

8339 > len(ms.InsertTasks[tasks.CategoryReplication]) > 0 {

8340 return softassert.UnexpectedInternalErr(

8341 ms.logger,

8345 }

8346

8347 > return nil mutable_state_impl.go

8348 }

8349

8350 > func (ms *MutableStateImpl) cleanupTransaction() error { mutable_state_impl.go

8351 > ms.updateActivityInfos = make(map[int64]*persistencespb.ActivityInfo)

8352 > ms.deleteActivityInfos = make(map[int64]struct{})

8353 > ms.syncActivityTasks = make(map[int64]struct{})

8354 >

8355 > ms.updateTimerInfos = make(map[string]*persistencespb.TimerInfo)

8356 > ms.deleteTimerInfos = make(map[string]struct{})

8357 >

8358 > ms.updateChildExecutionInfos = make(map[int64]*persistencespb.ChildExecutionInfo)

8359 > ms.deleteChildExecutionInfos = make(map[int64]struct{})

8360 >

8361 > ms.updateRequestCancelInfos = make(map[int64]*persistencespb.RequestCancelInfo)

8362 > ms.deleteRequestCancelInfos = make(map[int64]struct{})

8363 >

8364 > ms.updateSignalInfos = make(map[int64]*persistencespb.SignalInfo)

8365 > ms.deleteSignalInfos = make(map[int64]struct{})

8366 >

8367 > ms.updateSignalRequestedIDs = make(map[string]struct{})

8368 > ms.deleteSignalRequestedIDs = make(map[string]struct{})

8369 >

8370 > ms.visibilityUpdated = false

8371 > ms.executionStateUpdated = false

8372 > ms.workflowTaskUpdated = false

8373 > ms.isResetStateUpdated = false

8374 > ms.timeSkippingInfoUpdated = false

8375 > ms.updateInfoUpdated = make(map[string]struct{})

8376 > ms.timerInfosUserDataUpdated = make(map[string]struct{})

8377 > ms.activityInfosUserDataUpdated = make(map[int64]struct{})

8378 > ms.reapplyEventsCandidate = nil

8379 > ms.subStateMachineDeleted = false

8380 > ms.replayEventBatchID = common.EmptyEventID

8381 >

8382 > ms.stateInDB = ms.executionState.State

8383 > ms.nextEventIDInDB = ms.GetNextEventID()

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

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

8386 > } else { mutable_state_impl.go

8387 ms.versionedTransitionInDB = nil

8388 }

8389 // ms.dbRecordVersion remains the same

8390

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

8392 > ms.timeSource,

8393 > ms.shard.GenerateTaskIDs,

8394 > ms.GetCurrentVersion(),

8395 > ms.nextEventIDInDB,

8396 > ms.bufferEventsInDB,

8397 > ms.metricsHandler,

8398 > ms.config.MaximumEventBatchSizeInBytes,

8399 > )

8400 >

8401 > ms.InsertTasks = make(map[tasks.Category][]tasks.Task)

8402 > ms.BestEffortDeleteTasks = make(map[tasks.Category][]tasks.Key)

8403 >

8404 > // Clear outputs for the next transaction.

8405 > ms.stateMachineNode.ClearTransactionState()

8406 > // Clear out transient state machine state.

8407 > ms.currentTransactionAddedStateMachineEventTypes = nil

8408 >

8409 > return nil

8410 }

8411

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

8476 }

8477 }

8478

8479 > return workflowEventsSeq, newEventsBatches, newBufferBatch, clearBuffer, nil mutable_state_impl.go

8480 }

8481

8483 transactionPolicy historyi.TransactionPolicy,

8484 eventBatches [][]*historypb.HistoryEvent,

8485 > ) ([]tasks.Task, error) { mutable_state_impl.go

8486 > switch transactionPolicy {

8487 > case historyi.TransactionPolicyActive: mutable_state_impl.go

8488 > if ms.generateReplicationTask() {

8489 return ms.taskGenerator.GenerateHistoryReplicationTasks(eventBatches)

8490 }

8491 > return nil, nil mutable_state_impl.go

8492 case historyi.TransactionPolicyPassive:

8493 return nil, nil

8499 func (ms *MutableStateImpl) syncActivityToReplicationTask(

8500 transactionPolicy historyi.TransactionPolicy,

8501 > ) []tasks.Task { mutable_state_impl.go

8502 > now := time.Now().UTC()

8503 > switch transactionPolicy {

8504 > case historyi.TransactionPolicyActive: mutable_state_impl.go

8505 > if ms.generateReplicationTask() {

8506 var activityIDs map[int64]struct{}

8507 if ms.disablingTransitionHistory() {

8525 )

8526 }

8527 > return nil mutable_state_impl.go

8528 case historyi.TransactionPolicyPassive:

8529 return emptyTasks

8537 eventBatches [][]*historypb.HistoryEvent,

8538 clearBufferEvents bool,

8539 > ) []tasks.Task { mutable_state_impl.go

8540 > switch transactionPolicy {

8541 > case historyi.TransactionPolicyActive: mutable_state_impl.go

8542 > if !ms.generateReplicationTask() {

8543 > return emptyTasks mutable_state_impl.go

8544 > }

8545

8546 // Skip if there are no HSM children (no outbound tasks to generate)

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 { mutable_state_impl.go

8583 > activityInfo.StartedEventId = startedEventID mutable_state_impl.go

8584 > ms.updateActivityInfos[activityInfo.ScheduledEventId] = activityInfo

8585 > ms.activityInfosUserDataUpdated[activityInfo.ScheduledEventId] = struct{}{}

8586 > continue

8587 }

8588 if childInfo, ok := ms.GetChildExecutionInfo(scheduledEventID); ok {

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

8637 }

8638

8639 > return nil mutable_state_impl.go

8640 }

8641

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 mutable_state_impl.go

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 mutable_state_impl.go

8657 > }

8658

8659 // workflow close

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

8712 }

8717

8718 // Handling mutable state turn from standby to active, while having a workflow task on the fly

8719 > workflowTask := ms.GetStartedWorkflowTask() mutable_state_impl.go

8720 > currentVersion := ms.GetCurrentVersion()

8721 > if workflowTask == nil || workflowTask.Version >= currentVersion {

8722 > // no pending workflow tasks, no buffered events

8723 > // or workflow task has higher / equal version

8724 > return false, nil

8725 > }

8726

8727 lastEventVersion, err := ms.GetLastEventVersion()

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

8826 }

8827

8828 > func (ms *MutableStateImpl) BufferSizeAcceptable() bool { mutable_state_impl.go

8829 > if ms.hBuilder.NumBufferedEvents() > ms.config.MaximumBufferedEventsBatch() {

8830 return false

8831 }

8832

8833 > if ms.hBuilder.SizeInBytesOfBufferedEvents() > ms.config.MaximumBufferedEventsSizeInBytes() { mutable_state_impl.go

8834 return false

8835 }

8836 > return true mutable_state_impl.go

8837 }

8838

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

8845 }

8846

8847 > if ms.BufferSizeAcceptable() { mutable_state_impl.go

8848 > return nil mutable_state_impl.go

8849 > }

8850

8851 // Handling buffered events size issue

8870 func (ms *MutableStateImpl) closeTransactionHandleWorkflowResetTask(

8871 transactionPolicy historyi.TransactionPolicy,

8872 > ) error { mutable_state_impl.go

8873 > if transactionPolicy == historyi.TransactionPolicyPassive ||

8874 > !ms.IsWorkflowExecutionRunning() {

8875 return nil

8876 }

8877

8878 > namespaceEntry, err := ms.shard.GetNamespaceRegistry().GetNamespaceByID(namespace.ID(ms.executionInfo.NamespaceId)) mutable_state_impl.go

8879 > if err != nil {

8880 return err

8881 }

8882 > if _, pt := FindAutoResetPoint( mutable_state_impl.go

8883 > ms.timeSource,

8884 > namespaceEntry.VerifyBinaryChecksum,

8885 > ms.GetExecutionInfo().AutoResetPoints,

8886 > ); pt != nil {

8887 if err := ms.taskGenerator.GenerateWorkflowResetTasks(); err != nil {

8888 return err

8897 )

8898 }

8899 > return nil mutable_state_impl.go

8900 }

8901

8902 func (ms *MutableStateImpl) closeTransactionHandleActivityUserTimerTasks(

8903 transactionPolicy historyi.TransactionPolicy,

8904 > ) error { mutable_state_impl.go

8905 > switch transactionPolicy {

8906 > case historyi.TransactionPolicyActive: mutable_state_impl.go

8907 > if !ms.IsWorkflowExecutionRunning() {

8908 return nil

8909 }

8910 > if err := ms.taskGenerator.GenerateActivityTimerTasks(); err != nil { mutable_state_impl.go

8911 return err

8912 }

8913 > return ms.taskGenerator.GenerateUserTimerTasks() mutable_state_impl.go

8914 case historyi.TransactionPolicyPassive:

8915 return nil

8923 // Any other task type is preserved in order.

8924 // Eg: [START, UPSERT, TP1, CLOSE, TP2, TP3] -> [TP1, CLOSE, TP2, TP3]

8925 > func (ms *MutableStateImpl) closeTransactionCollapseVisibilityTasks() { mutable_state_impl.go

8926 > visTasks := ms.InsertTasks[tasks.CategoryVisibility]

8927 > if len(visTasks) < 2 {

8928 > return mutable_state_impl.go

8929 > }

8930 > var visTaskToKeep tasks.Task mutable_state_impl.go

8931 > lastIndex := -1

8932 > for i, task := range visTasks {

8933 > switch task.GetType() {

8934 case enumsspb.TASK_TYPE_VISIBILITY_START_EXECUTION,

8935 enumsspb.TASK_TYPE_VISIBILITY_UPSERT_EXECUTION,

8936 enumsspb.TASK_TYPE_VISIBILITY_CLOSE_EXECUTION,

8937 > enumsspb.TASK_TYPE_VISIBILITY_DELETE_EXECUTION: mutable_state_impl.go

8938 > if visTaskToKeep == nil || task.GetType() >= visTaskToKeep.GetType() {

8939 > visTaskToKeep = task

8940 > }

8941 > lastIndex = i

8942 }

8943 }

8944 > if visTaskToKeep == nil { mutable_state_impl.go

8945 return

8946 }

8947 > collapsedVisTasks := make([]tasks.Task, 0, len(visTasks)) mutable_state_impl.go

8948 > for i, task := range visTasks {

8949 > switch task.GetType() {

8950 case enumsspb.TASK_TYPE_VISIBILITY_START_EXECUTION,

8951 enumsspb.TASK_TYPE_VISIBILITY_UPSERT_EXECUTION,

8952 enumsspb.TASK_TYPE_VISIBILITY_CLOSE_EXECUTION,

8953 > enumsspb.TASK_TYPE_VISIBILITY_DELETE_EXECUTION: mutable_state_impl.go

8954 > if i == lastIndex {

8955 > collapsedVisTasks = append(collapsedVisTasks, visTaskToKeep)

8956 > }

8957 default:

8958 collapsedVisTasks = append(collapsedVisTasks, task)

8959 }

8960 }

8961 > ms.InsertTasks[tasks.CategoryVisibility] = collapsedVisTasks mutable_state_impl.go

8962 }

8963

8964 > func (ms *MutableStateImpl) generateReplicationTask() bool { mutable_state_impl.go

8965 > return len(ms.namespaceEntry.ClusterNames(ms.GetWorkflowKey().WorkflowID)) > 1

8966 > }

8967

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

8984 > func (ms *MutableStateImpl) generateChecksum() *persistencespb.Checksum { mutable_state_impl.go

8985 > if !ms.shouldGenerateChecksum() {

8986 > return nil

8987 > }

8988 csum, err := generateMutableStateChecksum(ms)

8989 if err != nil {

8994 }

8995

8996 > func (ms *MutableStateImpl) shouldGenerateChecksum() bool { mutable_state_impl.go

8997 > if ms.namespaceEntry == nil {

8998 return false

8999 }

9000 > return rand.Intn(100) < ms.config.MutableStateChecksumGenProbability(ms.namespaceEntry.Name().String()) mutable_state_impl.go

9001 }

9002

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

9843 }

9844

9845 > func (ms *MutableStateImpl) GetReapplyCandidateEvents() []*historypb.HistoryEvent { mutable_state_impl.go

9846 > return ms.reapplyEventsCandidate

9847 > }

9848

9849 func (ms *MutableStateImpl) IsSubStateMachineDeleted() bool {

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 }