runID string,
startTime time.Time,
namespaceName := namespaceEntry.Name().String()
logger = log.NewLazyLogger(logger, func() []tag.Tag {
return []tag.Tag{
tag.WorkflowNamespace(namespaceName),
Atlas › Test
Exact test identity: go.temporal.io/server/service/history/ndc/TestStateRebuilderSuite/TestRebuild
go.temporal.io/server/service/history/ndcTestStateRebuilderSuite/TestRebuildTestRebuildExpand a file to inspect source; the > gutter marks covered lines.
runID string,
startTime time.Time,
namespaceName := namespaceEntry.Name().String()
logger = log.NewLazyLogger(logger, func() []tag.Tag {
return []tag.Tag{
tag.WorkflowNamespace(namespaceName),
})
updateActivityInfos: make(map[int64]*persistencespb.ActivityInfo),
pendingActivityTimerHeartbeats: make(map[int64]time.Time),
pendingActivityInfoIDs: make(map[int64]*persistencespb.ActivityInfo),
pendingActivityIDToEventID: make(map[string]int64),
deleteActivityInfos: make(map[int64]struct{}),
syncActivityTasks: make(map[int64]struct{}),
pendingTimerInfoIDs: make(map[string]*persistencespb.TimerInfo),
pendingTimerEventIDToID: make(map[int64]string),
updateTimerInfos: make(map[string]*persistencespb.TimerInfo),
deleteTimerInfos: make(map[string]struct{}),
updateChildExecutionInfos: make(map[int64]*persistencespb.ChildExecutionInfo),
pendingChildExecutionInfoIDs: make(map[int64]*persistencespb.ChildExecutionInfo),
deleteChildExecutionInfos: make(map[int64]struct{}),
updateRequestCancelInfos: make(map[int64]*persistencespb.RequestCancelInfo),
pendingRequestCancelInfoIDs: make(map[int64]*persistencespb.RequestCancelInfo),
deleteRequestCancelInfos: make(map[int64]struct{}),
updateSignalInfos: make(map[int64]*persistencespb.SignalInfo),
pendingSignalInfoIDs: make(map[int64]*persistencespb.SignalInfo),
deleteSignalInfos: make(map[int64]struct{}),
updateSignalRequestedIDs: make(map[string]struct{}),
pendingSignalRequestedIDs: make(map[string]struct{}),
deleteSignalRequestedIDs: make(map[string]struct{}),
// This field will be initialized with a real chasm tree at the end of this function
// when feature flag is enabled.
chasmTree: &noopChasmTree{},
approximateSize: 0,
chasmNodeSizes: make(map[string]int),
totalTombstones: 0,
currentVersion: namespaceEntry.FailoverVersion(workflowID),
bufferEventsInDB: nil,
stateInDB: enumsspb.WORKFLOW_EXECUTION_STATE_VOID,
nextEventIDInDB: common.FirstEventID,
dbRecordVersion: 1,
namespaceEntry: namespaceEntry,
appliedEvents: make(map[string]struct{}),
InsertTasks: make(map[tasks.Category][]tasks.Task),
BestEffortDeleteTasks: make(map[tasks.Category][]tasks.Key),
transitionHistoryEnabled: shard.GetConfig().EnableTransitionHistory(namespaceName),
visibilityUpdated: false,
executionStateUpdated: false,
workflowTaskUpdated: false,
updateInfoUpdated: make(map[string]struct{}),
timerInfosUserDataUpdated: make(map[string]struct{}),
activityInfosUserDataUpdated: make(map[int64]struct{}),
reapplyEventsCandidate: []*historypb.HistoryEvent{},
QueryRegistry: NewQueryRegistry(),
shard: shard,
clusterMetadata: shard.GetClusterMetadata(),
eventsCache: eventsCache,
config: shard.GetConfig(),
timeSource: shard.GetTimeSource(),
logger: logger,
metricsHandler: shard.GetMetricsHandler().WithTags(metrics.OperationTag(metrics.WorkflowContextScope)),
endpointRegistry: shard.EndpointRegistry(),
}
s.executionInfo = &persistencespb.WorkflowExecutionInfo{
NamespaceId: namespaceEntry.ID().String(),
WorkflowId: workflowID,
WorkflowTaskVersion: common.EmptyVersion,
WorkflowTaskScheduledEventId: common.EmptyEventID,
WorkflowTaskStartedEventId: common.EmptyEventID,
WorkflowTaskRequestId: emptyUUID,
WorkflowTaskTimeout: timestamp.DurationFromSeconds(0),
WorkflowTaskAttempt: 1,
LastCompletedWorkflowTaskStartedEventId: common.EmptyEventID,
StartTime: timestamppb.New(startTime),
ExecutionTime: timestamppb.New(startTime),
VersionHistories: versionhistory.NewVersionHistories(&historyspb.VersionHistory{}),
ExecutionStats: &persistencespb.ExecutionStats{HistorySize: 0},
SubStateMachinesByType: make(map[string]*persistencespb.StateMachineMap),
}
s.executionInfo.TaskGenerationShardClockTimestamp = shard.CurrentVectorClock().GetClock()
s.approximateSize += s.executionInfo.Size()
s.executionState = &persistencespb.WorkflowExecutionState{
RunId: runID,
State: enumsspb.WORKFLOW_EXECUTION_STATE_CREATED,
Status: enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING,
StartTime: timestamppb.New(startTime),
RequestIds: make(map[string]*persistencespb.RequestIDInfo),
}
s.approximateSize += s.executionState.Size()
s.hBuilder = historybuilder.New(
s.timeSource,
s.shard.GenerateTaskIDs,
s.currentVersion,
common.FirstEventID,
s.bufferEventsInDB,
s.metricsHandler,
s.config.MaximumEventBatchSizeInBytes,
)
s.taskGenerator = GetTaskGeneratorProvider().NewTaskGenerator(shard, s)
s.workflowTaskManager = newWorkflowTaskStateMachine(s, s.metricsHandler)
s.mustInitHSM()
// TODO@time-skipping: support time skipping for chasm
if s.config.EnableChasm(namespaceName) {
s.chasmTree = chasm.NewEmptyTree(
shard.ChasmRegistry(),
}
s.wrapTimeSourceWithTimeSkipping()
}
}
}
if ms.executionInfo.SubStateMachinesByType == nil {
ms.executionInfo.SubStateMachinesByType = make(map[string]*persistencespb.StateMachineMap)
}
// Error only occurs if some initialization path forgets to register the workflow state machine.
stateMachineNode, err := hsm.NewRoot(ms.shard.StateMachineRegistry(), StateMachineType, ms, ms.executionInfo.SubStateMachinesByType, ms)
mutable_state_impl.go
if err != nil {
panic(err)
}
}
return ms.chasmTree.ArchetypeID() == chasm.WorkflowArchetypeID
}
return ms.stateMachineNode
}
return ms.chasmTree
}
// ChasmEnabled returns true if the mutable state has a real chasm tree.
}
return definition.NewWorkflowKey(
ms.executionInfo.NamespaceId,
ms.executionInfo.WorkflowId,
ms.executionState.RunId,
)
}
currentVersionHistory, err := versionhistory.GetCurrentVersionHistory(ms.executionInfo.VersionHistories)
if err != nil {
return nil, err
}
}
runTimeout *durationpb.Duration,
treeID string,
// NOTE: Unfortunately execution timeout and run timeout are not yet initialized into ms.executionInfo at this point.
// TODO: Consider explicitly initializing mutable state with these timeout parameters instead of passing them in.
workflowKey := ms.GetWorkflowKey()
archetypeID := ms.ChasmTree().ArchetypeID()
if archetypeID != chasm.WorkflowArchetypeID {
return softassert.UnexpectedInternalErr(
ms.logger,
}
if duration := ms.namespaceEntry.Retention(); duration > 0 {
retentionDuration = durationpb.New(duration)
}
initialBranchToken, err := ms.shard.GetExecutionManager().GetHistoryBranchUtil().NewHistoryBranch(
mutable_state_impl.go
workflowKey.NamespaceID,
workflowKey.WorkflowID,
workflowKey.RunID,
treeID,
nil,
[]*persistencespb.HistoryBranchRange{},
runTimeout.AsDuration(),
executionTimeout.AsDuration(),
retentionDuration.AsDuration(),
)
if err != nil {
return err
}
}
func (ms *MutableStateImpl) SetCurrentBranchToken(
branchToken []byte,
currentVersionHistory, err := versionhistory.GetCurrentVersionHistory(ms.executionInfo.VersionHistories)
if err != nil {
return err
}
versionhistory.SetVersionHistoryBranchToken(currentVersionHistory, branchToken)
mutable_state_impl.go
return nil
}
func (ms *MutableStateImpl) SetHistoryBuilder(hBuilder *historybuilder.HistoryBuilder) {
mutable_state_impl.go
ms.hBuilder = hBuilder
}
func (ms *MutableStateImpl) SetBaseWorkflow(
}
func (ms *MutableStateImpl) GetExecutionInfo() *persistencespb.WorkflowExecutionInfo {
mutable_state_impl.go
return ms.executionInfo
}
func (ms *MutableStateImpl) GetExecutionState() *persistencespb.WorkflowExecutionState {
mutable_state_impl.go
return ms.executionState
}
func (ms *MutableStateImpl) FlushBufferedEvents() {
version int64,
forceUpdate bool,
if ms.transitionHistoryEnabled && len(ms.executionInfo.TransitionHistory) != 0 {
// this make sure current version >= last write version
lastVersionedTransition := ms.CurrentVersionedTransition()
ms.currentVersion = lastVersionedTransition.NamespaceFailoverVersion
versionHistory, err := versionhistory.GetCurrentVersionHistory(ms.executionInfo.VersionHistories)
mutable_state_impl.go
if err != nil {
return err
}
versionHistoryItem, err := versionhistory.GetLastVersionHistoryItem(versionHistory)
if err != nil {
return err
}
}
}
}
ms.timeSource,
ms.shard.GenerateTaskIDs,
ms.currentVersion,
ms.nextEventIDInDB,
ms.bufferEventsInDB,
ms.metricsHandler,
ms.config.MaximumEventBatchSizeInBytes,
)
return nil
}
// TODO: can we always return ms.currentVersion here?
if ms.executionInfo.VersionHistories != nil {
return ms.currentVersion
}
if ms.transitionHistoryEnabled && len(ms.executionInfo.TransitionHistory) != 0 {
}
return ms.namespaceEntry
}
// AddHistoryEvent adds any history event to this workflow execution.
// GenerateEventLoadToken calls should reference. Only needed during replay. The rebuilder sets it
// for each batch before applying its events.
ms.replayEventBatchID = batchID
}
// GenerateEventLoadToken generates a token for loading a history event at a later time. The token encodes the event ID
}
ms.executionInfo.StickyTaskQueue = ""
ms.executionInfo.StickyScheduleToStartTimeout = nil
}
func (ms *MutableStateImpl) IsStickyTaskQueueSet() bool {
}
wType := &commonpb.WorkflowType{}
wType.Name = ms.executionInfo.WorkflowTypeName
return wType
}
func (ms *MutableStateImpl) GetQueryRegistry() historyi.QueryRegistry {
}
return ms.timeSource.Now()
}
// GetWorkflowCloseTime returns workflow closed time, returns a zero time for open workflow
func (ms *MutableStateImpl) writeEventToCache(
event *historypb.HistoryEvent,
// For start event: store it here so the recordWorkflowStarted transfer task doesn't need to
// load it from database.
// For completion event: store it here so we can communicate the result to parent execution
// during the processing of DeleteTransferTask without loading this event from database.
// For Update events: store it here so that Update disposition lookups can be fast.
ms.eventsCache.PutEvent(
events.EventKey{
NamespaceID: namespace.ID(ms.executionInfo.NamespaceId),
WorkflowID: ms.executionInfo.WorkflowId,
RunID: ms.executionState.RunId,
EventID: event.GetEventId(),
Version: event.GetVersion(),
},
event,
)
}
func (ms *MutableStateImpl) HasParentExecution() bool {
}
func (ms *MutableStateImpl) GetPendingActivityInfos() map[int64]*persistencespb.ActivityInfo {
mutable_state_impl.go
return ms.pendingActivityInfoIDs
}
func (ms *MutableStateImpl) GetPendingTimerInfos() map[string]*persistencespb.TimerInfo {
mutable_state_impl.go
return ms.pendingTimerInfoIDs
}
func (ms *MutableStateImpl) GetPendingChildExecutionInfos() map[int64]*persistencespb.ChildExecutionInfo {
}
return ms.workflowTaskManager.HasStartedWorkflowTask()
}
func (ms *MutableStateImpl) GetStartedWorkflowTask() *historyi.WorkflowTaskInfo {
// GetNextEventID returns next event ID
return ms.hBuilder.NextEventID()
}
// GetStartedEventIdForLastCompletedWorkflowTask returns last started workflow task event ID
}
switch ms.executionState.State {
return true
case enumsspb.WORKFLOW_EXECUTION_STATE_RUNNING:
return true
eventType enumspb.EventType,
eventID int64,
ms.approximateSize -= ms.executionState.Size()
if ms.executionState.RequestIds == nil {
ms.executionState.RequestIds = make(map[string]*persistencespb.RequestIDInfo, 1)
}
ms.executionState.RequestIds[requestID] = &persistencespb.RequestIDInfo{
mutable_state_impl.go
EventType: eventType,
EventId: eventID,
}
if eventType == enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED {
ms.executionState.CreateRequestId = requestID
}
ms.approximateSize += ms.executionState.Size()
}
requestID string,
startEvent *historypb.HistoryEvent,
if ms.executionInfo.NamespaceId != ms.namespaceEntry.ID().String() {
return serviceerror.NewInternalf("applying conflicting namespace ID: %v != %v",
ms.executionInfo.NamespaceId, ms.namespaceEntry.ID().String())
}
return serviceerror.NewInternalf("applying conflicting workflow ID: %v != %v",
ms.executionInfo.WorkflowId, execution.GetWorkflowId())
}
return serviceerror.NewInternalf("applying conflicting run ID: %v != %v",
ms.executionState.RunId, execution.GetRunId())
}
ms.AttachRequestID(requestID, startEvent.EventType, startEvent.EventId)
ms.approximateSize -= ms.executionInfo.Size()
ms.executionInfo.FirstExecutionRunId = event.GetFirstExecutionRunId()
ms.executionInfo.TaskQueue = event.TaskQueue.GetName()
ms.executionInfo.WorkflowTypeName = event.WorkflowType.GetName()
ms.executionInfo.WorkflowRunTimeout = event.GetWorkflowRunTimeout()
ms.executionInfo.WorkflowExecutionTimeout = event.GetWorkflowExecutionTimeout()
ms.executionInfo.DefaultWorkflowTaskTimeout = event.GetWorkflowTaskTimeout()
ms.executionInfo.OriginalExecutionRunId = event.GetOriginalExecutionRunId()
ms.approximateSize -= ms.executionState.Size()
ms.executionState.FirstExecutionRunId = event.GetFirstExecutionRunId()
if err := ms.addCompletionCallbacks(
startEvent,
requestID,
event.GetCompletionCallbacks(),
); err != nil {
return err
}
enumsspb.WORKFLOW_EXECUTION_STATE_CREATED,
enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING,
); err != nil {
return err
}
ms.executionInfo.LastCompletedWorkflowTaskStartedEventId = common.EmptyEventID
mutable_state_impl.go
ms.executionInfo.LastFirstEventId = startEvent.GetEventId()
ms.executionInfo.WorkflowTaskVersion = common.EmptyVersion
ms.executionInfo.WorkflowTaskScheduledEventId = common.EmptyEventID
ms.executionInfo.WorkflowTaskStartedEventId = common.EmptyEventID
ms.executionInfo.WorkflowTaskRequestId = emptyUUID
ms.executionInfo.WorkflowTaskTimeout = timestamp.DurationFromSeconds(0)
ms.executionInfo.CronSchedule = event.GetCronSchedule()
if event.ParentWorkflowExecution != nil {
ms.executionInfo.ParentNamespaceId = event.GetParentWorkflowNamespaceId()
ms.executionInfo.ParentWorkflowId = event.ParentWorkflowExecution.GetWorkflowId()
}
ms.executionInfo.ParentInitiatedId = event.GetParentInitiatedEventId()
}
ms.executionInfo.ParentInitiatedVersion = event.GetParentInitiatedEventVersion()
}
ms.executionInfo.RootWorkflowId = event.RootWorkflowExecution.GetWorkflowId()
ms.executionInfo.RootRunId = event.RootWorkflowExecution.GetRunId()
ms.executionInfo.RootWorkflowId = execution.GetWorkflowId()
ms.executionInfo.RootRunId = execution.GetRunId()
}
// todo@time-skipping: apply time skipping to WorkflowStartDelay
ms.executionState.StartTime.AsTime().Add(event.GetFirstWorkflowTaskBackoff().AsDuration()),
)
ms.executionInfo.Attempt = event.GetAttempt()
if !timestamp.TimeValue(event.GetWorkflowExecutionExpirationTime()).IsZero() {
ms.executionInfo.WorkflowExecutionExpirationTime = event.GetWorkflowExecutionExpirationTime()
}
workflowRunTimeoutDuration := ms.executionInfo.WorkflowRunTimeout.AsDuration()
// if workflowRunTimeoutDuration == 0 then the workflowRunTimeoutTime will be 0
// meaning that there is not workflow run timeout
if workflowRunTimeoutDuration != 0 {
firstWorkflowTaskDelayDuration := event.GetFirstWorkflowTaskBackoff().AsDuration()
mutable_state_impl.go
workflowRunTimeoutDuration = workflowRunTimeoutDuration + firstWorkflowTaskDelayDuration
workflowRunTimeoutTime = ms.executionState.StartTime.AsTime().Add(workflowRunTimeoutDuration)
workflowExecutionTimeoutTime := timestamp.TimeValue(ms.executionInfo.WorkflowExecutionExpirationTime)
if !workflowExecutionTimeoutTime.IsZero() && workflowRunTimeoutTime.After(workflowExecutionTimeoutTime) {
workflowRunTimeoutTime = workflowExecutionTimeoutTime
}
}
ms.executionInfo.WorkflowRunExpirationTime = timestamppb.New(workflowRunTimeoutTime)
mutable_state_impl.go
if event.RetryPolicy != nil {
ms.executionInfo.HasRetryPolicy = true
ms.executionInfo.RetryBackoffCoefficient = event.RetryPolicy.GetBackoffCoefficient()
}
ms.executionInfo.AutoResetPoints = rolloverAutoResetPointsWithExpiringTime(
mutable_state_impl.go
event.GetPrevAutoResetPoints(),
event.GetContinuedExecutionRunId(),
timestamp.TimeValue(startEvent.GetEventTime()),
ms.namespaceEntry.Retention(),
)
if event.Memo != nil {
ms.executionInfo.Memo = event.Memo.GetFields()
}
ms.executionInfo.SearchAttributes = event.SearchAttributes.GetIndexedFields()
}
if ms.executionInfo.VersioningInfo == nil {
ms.executionInfo.VersioningInfo = &workflowpb.WorkflowExecutionVersioningInfo{}
}
if ms.executionInfo.VersioningInfo == nil {
ms.executionInfo.VersioningInfo = &workflowpb.WorkflowExecutionVersioningInfo{}
// target version upgrade from the started event. This is the same public API
// type, so no conversion needed.
if event.GetContinuedExecutionRunId() != "" && event.GetInheritedPinnedVersion() != nil {
mutable_state_impl.go
ms.executionInfo.DeclinedTargetVersionUpgrade = event.GetDeclinedTargetVersionUpgrade()
}
// Populate the versioningInfo if the inheritedAutoUpgradeInfo is present.
ms.SetVersioningRevisionNumber(event.GetInheritedAutoUpgradeInfo().GetSourceDeploymentRevisionNumber())
// TODO (Shivam): Remove this once you make SetDeploymentVersion and SetVersioningBehavior methods with nil checks
}
if inheritedBuildId := event.InheritedBuildId; inheritedBuildId != "" {
mutable_state_impl.go
ms.executionInfo.InheritedBuildId = inheritedBuildId
if err := ms.UpdateBuildIdAssignment(inheritedBuildId); err != nil {
return err
}
} else if event.SourceVersionStamp.GetUseVersioning() && event.SourceVersionStamp.GetBuildId() != "" ||
mutable_state_impl.go
ms.GetEffectiveVersioningBehavior() != enumspb.VERSIONING_BEHAVIOR_UNSPECIFIED {
// TODO: [cleanup-old-wv]
limit := ms.config.SearchAttributesSizeOfValueLimit(string(ms.namespaceEntry.Name()))
// This will include override and inheritance, but not transition, because WF never starts with a transition
ms.executionInfo.WorkerDeploymentName = ms.GetEffectiveDeployment().GetSeriesName()
mutable_state_impl.go
if inheritedBuildId := event.InheritedBuildId; inheritedBuildId != "" {
ms.executionInfo.InheritedBuildId = inheritedBuildId
if err := ms.UpdateBuildIdAssignment(inheritedBuildId); err != nil {
}
ms.executionInfo.MostRecentWorkerVersionStamp = event.SourceVersionStamp
mutable_state_impl.go
ms.executionInfo.Priority = event.Priority
if tsc, stateProp := event.GetTimeSkippingConfig(), event.GetTimeSkippingStatePropagation(); tsc != nil || stateProp.GetInitialSkippedDuration().AsDuration() > 0 {
if err := ms.initTimeSkippingInfo(tsc, stateProp); err != nil {
return err
}
ms.approximateSize += ms.executionState.Size()
ms.writeEventToCache(startEvent)
return nil
}
requestID string,
completionCallbacks []*commonpb.Callback,
if len(completionCallbacks) == 0 {
}
if ms.chasmCallbacksEnabled() {
// Initialize chasm tree once for new workflows.
func (ms *MutableStateImpl) ApplyWorkflowExecutionSignaled(
event *historypb.HistoryEvent,
// Increment signal count in mutable state for this workflow execution
ms.executionInfo.SignalCount++
// Add signal requestID to workflow CHASM tree (if feature is enabled)
signalEventAttrs, ok := event.GetAttributes().(*historypb.HistoryEvent_WorkflowExecutionSignaledEventAttributes)
if !ok {
return softassert.UnexpectedInternalErr(
ms.logger,
)
}
requestID := signalEventAttrs.WorkflowExecutionSignaledEventAttributes.GetRequestId()
mutable_state_impl.go
if requestID != "" && ms.ChasmSignalBacklinksEnabled() {
ctx := context.Background()
ms.EnsureChasmWorkflowComponent(ctx)
metrics.ChasmIncomingSignalWritten.With(ms.metricsHandler.WithTags(nsTag)).Record(1)
}
}
newExecutionStartTime time.Time,
namespaceRetention time.Duration,
if resetPoints.GetPoints() == nil {
return resetPoints
}
newPoints := make([]*workflowpb.ResetPointInfo, 0, len(resetPoints.Points))
// For continue-as-new, new execution start time is the same as previous execution close time,
}
if ms.executionInfo.ExecutionStats == nil {
ms.executionInfo.ExecutionStats = &persistencespb.ExecutionStats{}
}
}
}
if ms.executionInfo.ExecutionStats == nil {
ms.executionInfo.ExecutionStats = &persistencespb.ExecutionStats{}
}
}
func (ms *MutableStateImpl) AddTasks(
newTasks ...tasks.Task,
now := ms.Now()
for _, task := range newTasks {
if chasmTask, ok := task.(*tasks.ChasmTask); ok &&
chasmTask.GetCategory() == tasks.CategoryVisibility &&
ms.stateInDB == enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED {
softassert.Fail(ms.logger, "CHASM visibility task added on already-closed execution")
}
// Drop tasks scheduled too far in the future. VisibilityTime hasn't been
// shifted to wall-clock yet (the conversion runs below), so both sides are
// virtual here; the difference is frame-invariant (skip cancels). Keep
// `now` from ms.Now() so both stay in the same frame.
if category.Type() == tasks.CategoryTypeScheduled &&
task.GetVisibilityTime().Sub(now) > maxScheduledTaskDuration {
ms.logger.Info("Dropped long duration scheduled task.", tasks.Tags(task)...)
continue
// vs. real distinction. The CategoryTypeScheduled drop-check above runs first so it
// compares virtual-vs-virtual (now is also virtual).
}
ms.chasmPureTasks = append(ms.chasmPureTasks, chasmPureTask)
maxPureTasks := ms.config.ChasmMaxInMemoryPureTasks()
}
}
}
state enumsspb.WorkflowExecutionState,
status enumspb.WorkflowExecutionStatus,
if state == ms.executionState.State && status == ms.executionState.Status {
}
if state != enumsspb.WORKFLOW_EXECUTION_STATE_ZOMBIE &&
ms.executionState.State != enumsspb.WORKFLOW_EXECUTION_STATE_ZOMBIE {
// isStateDirty is used upon closing transaction to determine if application data has been updated, and
// mutable state should move to a new versioned transition.
// TODO: we need to track more workflow state changes
// e.g. changes to executionInfo.CancelRequested
// They are mostly covered by history builder check today.
return ms.hBuilder.IsDirty() ||
len(ms.activityInfosUserDataUpdated) > 0 ||
len(ms.deleteActivityInfos) > 0 ||
len(ms.timerInfosUserDataUpdated) > 0 ||
len(ms.deleteTimerInfos) > 0 ||
len(ms.updateChildExecutionInfos) > 0 ||
len(ms.deleteChildExecutionInfos) > 0 ||
len(ms.updateRequestCancelInfos) > 0 ||
len(ms.deleteRequestCancelInfos) > 0 ||
len(ms.updateSignalInfos) > 0 ||
len(ms.deleteSignalInfos) > 0 ||
len(ms.updateSignalRequestedIDs) > 0 ||
len(ms.deleteSignalRequestedIDs) > 0 ||
len(ms.updateInfoUpdated) > 0 ||
ms.visibilityUpdated ||
ms.executionStateUpdated ||
ms.workflowTaskUpdated ||
(ms.stateMachineNode != nil && ms.stateMachineNode.Dirty()) ||
ms.chasmTree.IsStateDirty() ||
ms.isResetStateUpdated ||
ms.timeSkippingInfoUpdated
}
func (ms *MutableStateImpl) IsTransitionHistoryEnabled() bool {
ctx context.Context,
transactionPolicy historyi.TransactionPolicy,
) (*persistence.WorkflowSnapshot, []*persistence.WorkflowEvents, error) {
mutable_state_impl.go
result, err := ms.closeTransaction(ctx, transactionPolicy)
if err != nil {
return nil, nil, err
}
// TODO do we need the functionality to generate snapshot with buffered events?
return nil, nil, softassert.UnexpectedInternalErr(
}
ExecutionInfo: ms.executionInfo,
ExecutionState: ms.executionState,
NextEventID: ms.hBuilder.NextEventID(),
ActivityInfos: ms.pendingActivityInfoIDs,
TimerInfos: ms.pendingTimerInfoIDs,
ChildExecutionInfos: ms.pendingChildExecutionInfoIDs,
RequestCancelInfos: ms.pendingRequestCancelInfoIDs,
SignalInfos: ms.pendingSignalInfoIDs,
SignalRequestedIDs: ms.pendingSignalRequestedIDs,
ChasmNodes: ms.chasmTree.Snapshot(nil).Nodes,
Tasks: ms.InsertTasks,
Condition: ms.nextEventIDInDB,
DBRecordVersion: ms.dbRecordVersion,
Checksum: result.checksum,
}
ms.checksum = result.checksum
if err := ms.cleanupTransaction(); err != nil {
return nil, nil, err
}
}
func (ms *MutableStateImpl) SetContextMetadata(
ctx context.Context,
switch ms.chasmTree.ArchetypeID() {
// Set workflow type
if wfType := ms.GetWorkflowType(); wfType != nil && wfType.GetName() != "" {
contextutil.ContextMetadataSet(ctx, contextutil.MetadataKeyWorkflowType, wfType.GetName())
mutable_state_impl.go
}
// Set workflow task queue
contextutil.ContextMetadataSet(ctx, contextutil.MetadataKeyWorkflowTaskQueue, ms.executionInfo.TaskQueue)
mutable_state_impl.go
}
for _, activityID := range contextutil.ContextMetadataGetMarkedActivityIDs(ctx) {
mutable_state_impl.go
if ai, ok := ms.GetActivityByActivityID(activityID); ok {
contextutil.ContextMetadataSet(ctx, contextutil.ActivityTypeKey(ai.ScheduledEventId), ai.ActivityType.GetName())
ctx context.Context,
transactionPolicy historyi.TransactionPolicy,
ms.SetContextMetadata(ctx)
if err := ms.closeTransactionWithPolicyCheck(
transactionPolicy,
); err != nil {
return closeTransactionResult{}, err
}
transactionPolicy,
); err != nil {
return closeTransactionResult{}, err
}
// and need to reconsider the sequence of time skipping close trx handling in this function
// when supporting chasm.
regenTimerTasksForWorkflowTimeSkipping := ms.closeTransactionHandleWorkflowTimeSkipping(ctx, transactionPolicy)
mutable_state_impl.go
// Save if the state is dirty before closeTransactionPrepareEvents since it flushes the buffer
// events, and therefore change the dirty state.
isStateDirty := ms.isStateDirty()
// closeTransactionPrepareEvents must be called after closeTransactionHandleWorkflowTask because
// the latter might fail the workflow task and buffered events must be flushed afterwards.
// We need to save the value of ms.isStateDirty() before calling closeTransactionPrepareEvents
// because flushing the buffered events might change the dirty state.
workflowEventsSeq, eventBatches, bufferEvents, clearBuffer, err := ms.closeTransactionPrepareEvents(transactionPolicy)
if err != nil {
return closeTransactionResult{}, err
}
// cluster — standby (passive) replays events that were already stamped by
// the active side, and we must not overwrite those principals.
principal := headers.GetPrincipal(ctx)
for _, we := range workflowEventsSeq {
// CloseTransaction() on chasmTree may update execution state & status,
// so must be called before closeTransactionUpdateTransitionHistory().
if err != nil {
return closeTransactionResult{}, err
}
if ms.closeTransactionShouldSkipPersistence(isStateDirty, chasmNodesMutation) {
mutable_state_impl.go
return closeTransactionResult{
skipPersistence: true,
}
ms.approximateSize -= ms.chasmNodeSizes[nodePath]
delete(ms.chasmNodeSizes, nodePath)
}
newSize := len(nodePath) + node.Size()
ms.approximateSize += newSize - ms.chasmNodeSizes[nodePath]
}
transactionPolicy,
); err != nil {
return closeTransactionResult{}, err
}
ms.closeTransactionUpdateLastRunningClock(transactionPolicy, workflowEventsSeq)
}
// todo@TimeSkipping, we can move update versioned transition to inside closeTransactionHandleWorkflowTimeSkipping
transactionPolicy,
)
ms.closeTransactionTrackTombstones(transactionPolicy, chasmNodesMutation)
// generate tasks
if err := ms.closeTransactionPrepareTasks(
transactionPolicy,
eventBatches,
clearBuffer,
regenTimerTasksForWorkflowTimeSkipping,
); err != nil {
return closeTransactionResult{}, err
}
ms.executionInfo.LastUpdateTime = timestamppb.New(ms.timeSource.Now())
// We generate checksum here based on the assumption that the returned
// snapshot object is considered immutable. As of this writing, the only
// code that modifies the returned object lives inside Context.resetWorkflowExecution.
// Currently, the updates done inside Context.resetWorkflowExecution don't
// impact the checksum calculation.
checksum := ms.generateChecksum()
if ms.dbRecordVersion == 0 {
// noop, existing behavior
}
workflowEventsSeq: workflowEventsSeq,
bufferEvents: bufferEvents,
clearBuffer: clearBuffer,
checksum: checksum,
chasmNodesMutation: chasmNodesMutation,
}, nil
}
func (ms *MutableStateImpl) closeTransactionShouldSkipPersistence(isStateDirty bool, chasmNodesMutation chasm.NodesMutation) bool {
mutable_state_impl.go
return !ms.IsWorkflow() && !isStateDirty && chasmNodesMutation.IsEmpty()
}
func (ms *MutableStateImpl) closeTransactionHandleWorkflowTask(
transactionPolicy historyi.TransactionPolicy,
if err := ms.closeTransactionHandleBufferedEventsLimit(
transactionPolicy,
); err != nil {
return err
}
transactionPolicy,
); err != nil {
return err
}
return ms.closeTransactionHandleSpeculativeWorkflowTask(transactionPolicy)
mutable_state_impl.go
}
func (ms *MutableStateImpl) closeTransactionHandleWorkflowTaskScheduling(
transactionPolicy historyi.TransactionPolicy,
if transactionPolicy == historyi.TransactionPolicyPassive ||
!ms.IsWorkflowExecutionRunning() {
}
for _, t := range ms.currentTransactionAddedStateMachineEventTypes {
func (ms *MutableStateImpl) closeTransactionHandleSpeculativeWorkflowTask(
transactionPolicy historyi.TransactionPolicy,
if transactionPolicy == historyi.TransactionPolicyPassive ||
!ms.IsWorkflowExecutionRunning() {
}
// It is important to convert speculative WT to normal before prepareEventsAndReplicationTasks,
func (ms *MutableStateImpl) closeTransactionUpdateTransitionHistory(
transactionPolicy historyi.TransactionPolicy,
if transactionPolicy != historyi.TransactionPolicyActive {
// TODO: replication/standby logic will need a different way for updating transition history
mutable_state_impl.go
// when not syncing mutable state
return nil
}
if !ms.transitionHistoryEnabled {
func (ms *MutableStateImpl) closeTransactionTrackLastUpdateVersionedTransition(
transactionPolicy historyi.TransactionPolicy,
if transactionPolicy != historyi.TransactionPolicyActive {
// TODO: replication/standby logic will need a different way for updating LastUpdatedVersionedTransition
mutable_state_impl.go
// when reapplying history, especially when history replication tasks got batched.
return
}
if !ms.transitionHistoryEnabled {
}
func (ms *MutableStateImpl) closeTransactionHandleUnknownVersionedTransition() {
mutable_state_impl.go
if len(ms.executionInfo.TransitionHistory) != 0 {
if transitionhistory.Compare(
ms.versionedTransitionInDB,
// State changed but transition history not updated.
// We are in unknown versioned transition state, clear the transition history.
ms.executionInfo.PreviousTransitionHistory = ms.executionInfo.TransitionHistory
ms.executionInfo.LastTransitionHistoryBreakPoint = transitionhistory.CopyVersionedTransition(ms.CurrentVersionedTransition())
}
ms.executionInfo.SubStateMachineTombstoneBatches = nil
ms.totalTombstones = 0
for _, activityInfo := range ms.updateActivityInfos {
activityInfo.LastUpdateVersionedTransition = nil
}
timerInfo.LastUpdateVersionedTransition = nil
}
childInfo.LastUpdateVersionedTransition = nil
}
requestCancelInfo.LastUpdateVersionedTransition = nil
}
signalInfo.LastUpdateVersionedTransition = nil
}
ms.executionInfo.UpdateInfos[updateID].LastUpdateVersionedTransition = nil
}
if len(ms.updateSignalRequestedIDs) > 0 || len(ms.deleteSignalRequestedIDs) > 0 {
mutable_state_impl.go
ms.executionInfo.SignalRequestIdsLastUpdateVersionedTransition = nil
}
ms.executionInfo.VisibilityLastUpdateVersionedTransition = nil
}
ms.executionState.LastUpdateVersionedTransition = nil
}
ms.executionInfo.WorkflowTaskLastUpdateVersionedTransition = nil
}
// the error must be nil here since the fn passed into Walk() always returns nil
_ = ms.stateMachineNode.Walk(func(node *hsm.Node) error {
persistenceRepr := node.InternalRepr()
persistenceRepr.LastUpdateVersionedTransition.TransitionCount = 0
persistenceRepr.InitialVersionedTransition.TransitionCount = 0
return nil
})
}
}
transactionPolicy historyi.TransactionPolicy,
workflowEventsSeq []*persistence.WorkflowEvents,
if transactionPolicy != historyi.TransactionPolicyActive {
}
// Events can only be generated while mutable state is running,
transactionPolicy historyi.TransactionPolicy,
chasmNodesMutation chasm.NodesMutation,
if transactionPolicy != historyi.TransactionPolicyActive {
// Passive/Replication logic will update tombstone list when applying mutable state
mutable_state_impl.go
// snapshot or mutation.
return
}
if !ms.transitionHistoryEnabled {
clearBufferEvents bool,
regenerateTimerTasksForTimeSkipping bool,
if err := ms.closeTransactionHandleWorkflowResetTask(
transactionPolicy,
); err != nil {
return err
}
if err := ms.taskGenerator.GenerateDirtySubStateMachineTasks(ms.shard.StateMachineRegistry()); err != nil {
mutable_state_impl.go
return err
}
if err := ms.closeTransactionGenerateChasmRetentionTask(transactionPolicy); err != nil {
return err
}
// regardless of how many activity & user timer created
// so the calculation must be at the very end
if err := ms.closeTransactionHandleActivityUserTimerTasks(transactionPolicy); err != nil {
mutable_state_impl.go
return err
}
if err := ms.closeTransactionRegenTimerTasksForWorkflowTimeSkipping(transactionPolicy); err != nil {
return err
}
return ms.closeTransactionPrepareReplicationTasks(transactionPolicy, eventBatches, clearBufferEvents)
mutable_state_impl.go
}
func (ms *MutableStateImpl) closeTransactionGenerateChasmRetentionTask(
transactionPolicy historyi.TransactionPolicy,
if ms.IsWorkflow() ||
ms.executionState.State != enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED ||
ms.stateInDB == enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED {
}
// Generate retention timer for chasm executions if it's currentely completed
eventBatches [][]*historypb.HistoryEvent,
clearBufferEvents bool,
var replicationTasks []tasks.Task
if ms.config.ReplicationMultipleBatches() {
task, err := ms.eventsToReplicationTask(transactionPolicy, eventBatches)
if err != nil {
}
replicationTasks = append(replicationTasks, task...)
for _, historyEvents := range eventBatches {
task, err := ms.eventsToReplicationTask(transactionPolicy, [][]*historypb.HistoryEvent{historyEvents})
mutable_state_impl.go
if err != nil {
return err
}
}
}
replicationTasks = append(replicationTasks, ms.syncActivityToReplicationTask(transactionPolicy)...)
mutable_state_impl.go
replicationTasks = append(replicationTasks, ms.dirtyHSMToReplicationTask(transactionPolicy, eventBatches, clearBufferEvents)...)
archetypeID := ms.ChasmTree().ArchetypeID()
isWorkflow := archetypeID == chasm.WorkflowArchetypeID
if !isWorkflow && len(replicationTasks) != 0 {
return softassert.UnexpectedInternalErr(ms.logger, "chasm execution generated workflow replication tasks", nil)
}
case historyi.TransactionPolicyActive:
if ms.generateReplicationTask() {
}
}
default:
panic(fmt.Sprintf("unknown transaction policy: %v", transactionPolicy))
}
len(ms.InsertTasks[tasks.CategoryReplication]) > 0 {
return softassert.UnexpectedInternalErr(
ms.logger,
}
}
ms.updateActivityInfos = make(map[int64]*persistencespb.ActivityInfo)
ms.deleteActivityInfos = make(map[int64]struct{})
ms.syncActivityTasks = make(map[int64]struct{})
ms.updateTimerInfos = make(map[string]*persistencespb.TimerInfo)
ms.deleteTimerInfos = make(map[string]struct{})
ms.updateChildExecutionInfos = make(map[int64]*persistencespb.ChildExecutionInfo)
ms.deleteChildExecutionInfos = make(map[int64]struct{})
ms.updateRequestCancelInfos = make(map[int64]*persistencespb.RequestCancelInfo)
ms.deleteRequestCancelInfos = make(map[int64]struct{})
ms.updateSignalInfos = make(map[int64]*persistencespb.SignalInfo)
ms.deleteSignalInfos = make(map[int64]struct{})
ms.updateSignalRequestedIDs = make(map[string]struct{})
ms.deleteSignalRequestedIDs = make(map[string]struct{})
ms.visibilityUpdated = false
ms.executionStateUpdated = false
ms.workflowTaskUpdated = false
ms.isResetStateUpdated = false
ms.timeSkippingInfoUpdated = false
ms.updateInfoUpdated = make(map[string]struct{})
ms.timerInfosUserDataUpdated = make(map[string]struct{})
ms.activityInfosUserDataUpdated = make(map[int64]struct{})
ms.reapplyEventsCandidate = nil
ms.subStateMachineDeleted = false
ms.replayEventBatchID = common.EmptyEventID
ms.stateInDB = ms.executionState.State
ms.nextEventIDInDB = ms.GetNextEventID()
if len(ms.executionInfo.TransitionHistory) != 0 {
ms.versionedTransitionInDB = ms.CurrentVersionedTransition()
}
// ms.dbRecordVersion remains the same
ms.timeSource,
ms.shard.GenerateTaskIDs,
ms.GetCurrentVersion(),
ms.nextEventIDInDB,
ms.bufferEventsInDB,
ms.metricsHandler,
ms.config.MaximumEventBatchSizeInBytes,
)
ms.InsertTasks = make(map[tasks.Category][]tasks.Task)
ms.BestEffortDeleteTasks = make(map[tasks.Category][]tasks.Key)
// Clear outputs for the next transaction.
ms.stateMachineNode.ClearTransactionState()
// Clear out transient state machine state.
ms.currentTransactionAddedStateMachineEventTypes = nil
return nil
}
func (ms *MutableStateImpl) closeTransactionPrepareEvents(
transactionPolicy historyi.TransactionPolicy,
) ([]*persistence.WorkflowEvents, [][]*historypb.HistoryEvent, []*historypb.HistoryEvent, bool, error) {
mutable_state_impl.go
currentBranchToken, err := ms.GetCurrentBranchToken()
if err != nil {
return nil, nil, nil, false, err
}
historyMutation, err := ms.hBuilder.Finish(!ms.HasStartedWorkflowTask())
mutable_state_impl.go
if err != nil {
return nil, nil, nil, false, err
}
// TODO @wxing1292 need more refactoring to make the logic clean
newBufferBatch := historyMutation.DBBufferBatch
clearBuffer := historyMutation.DBClearBuffer
newEventsBatches := historyMutation.DBEventsBatches
ms.updatePendingEventIDs(historyMutation.ScheduledIDToStartedID, historyMutation.RequestIDToEventID)
workflowEventsSeq := make([]*persistence.WorkflowEvents, len(newEventsBatches))
historyNodeTxnIDs, err := ms.shard.GenerateTaskIDs(len(newEventsBatches))
if err != nil {
return nil, nil, nil, false, err
}
NamespaceID: ms.executionInfo.NamespaceId,
WorkflowID: ms.executionInfo.WorkflowId,
RunID: ms.executionState.RunId,
BranchToken: currentBranchToken,
PrevTxnID: ms.executionInfo.LastFirstEventTxnId,
TxnID: historyNodeTxnIDs[index],
Events: eventBatch,
}
ms.executionInfo.LastFirstEventId = eventBatch[0].GetEventId()
ms.executionInfo.LastFirstEventTxnId = historyNodeTxnIDs[index]
// Calculate and add the external payload size and count for this batch
if ms.config.ExternalPayloadsEnabled(ms.GetNamespaceEntry().Name().String()) {
externalPayloadSize, externalPayloadCount, err := CalculateExternalPayloadSize(eventBatch, ms.metricsHandler)
if err != nil {
return nil, nil, nil, false, err
}
ms.AddExternalPayloadCount(externalPayloadCount)
}
}
transactionPolicy,
workflowEventsSeq,
); err != nil {
return nil, nil, nil, false, err
}
lastEvent := lastEvents[len(lastEvents)-1]
if err := ms.updateWithLastWriteEvent(
lastEvent,
transactionPolicy,
); err != nil {
return nil, nil, nil, false, err
}
}
return workflowEventsSeq, newEventsBatches, newBufferBatch, clearBuffer, nil
mutable_state_impl.go
}
transactionPolicy historyi.TransactionPolicy,
eventBatches [][]*historypb.HistoryEvent,
switch transactionPolicy {
case historyi.TransactionPolicyActive:
if ms.generateReplicationTask() {
}
return nil, nil
return nil, nil
default:
panic(fmt.Sprintf("unknown transaction policy: %v", transactionPolicy))
func (ms *MutableStateImpl) syncActivityToReplicationTask(
transactionPolicy historyi.TransactionPolicy,
now := time.Now().UTC()
switch transactionPolicy {
case historyi.TransactionPolicyActive:
if ms.generateReplicationTask() {
}
return nil
return emptyTasks
default:
panic(fmt.Sprintf("unknown transaction policy: %v", transactionPolicy))
eventBatches [][]*historypb.HistoryEvent,
clearBufferEvents bool,
switch transactionPolicy {
case historyi.TransactionPolicyActive:
if !ms.generateReplicationTask() {
return emptyTasks
return emptyTasks
default:
panic(fmt.Sprintf("unknown transaction policy: %v", transactionPolicy))
scheduledIDToStartedID map[int64]int64,
requestIDToEventID map[string]int64,
for scheduledEventID, startedEventID := range scheduledIDToStartedID {
if activityInfo, ok := ms.GetActivityInfo(scheduledEventID); ok {
activityInfo.StartedEventId = startedEventID
}
}
var wf *chasmworkflow.Workflow
var chasmCtx chasm.MutableContext
lastEvent *historypb.HistoryEvent,
transactionPolicy historyi.TransactionPolicy,
if transactionPolicy == historyi.TransactionPolicyPassive {
return nil
}
currentVersionHistory, err := versionhistory.GetCurrentVersionHistory(ms.executionInfo.VersionHistories)
transactionPolicy historyi.TransactionPolicy,
workflowEventSeq []*persistence.WorkflowEvents,
if transactionPolicy == historyi.TransactionPolicyPassive ||
len(workflowEventSeq) == 0 {
}
// only do check if workflow is finished
func (ms *MutableStateImpl) closeTransactionWithPolicyCheck(
transactionPolicy historyi.TransactionPolicy,
switch transactionPolicy {
case historyi.TransactionPolicyActive:
// Cannot use ms.namespaceEntry.ActiveClusterName() because currentVersion may be updated during this transaction in
}
return nil
return nil
default:
panic(fmt.Sprintf("unknown transaction policy: %v", transactionPolicy))
func (ms *MutableStateImpl) closeTransactionHandleBufferedEventsLimit(
transactionPolicy historyi.TransactionPolicy,
if transactionPolicy == historyi.TransactionPolicyPassive ||
!ms.IsWorkflowExecutionRunning() {
}
if ms.BufferSizeAcceptable() {
func (ms *MutableStateImpl) closeTransactionHandleWorkflowResetTask(
transactionPolicy historyi.TransactionPolicy,
if transactionPolicy == historyi.TransactionPolicyPassive ||
!ms.IsWorkflowExecutionRunning() {
}
namespaceEntry, err := ms.shard.GetNamespaceRegistry().GetNamespaceByID(namespace.ID(ms.executionInfo.NamespaceId))
func (ms *MutableStateImpl) closeTransactionHandleActivityUserTimerTasks(
transactionPolicy historyi.TransactionPolicy,
switch transactionPolicy {
case historyi.TransactionPolicyActive:
if !ms.IsWorkflowExecutionRunning() {
}
return ms.taskGenerator.GenerateUserTimerTasks()
return nil
default:
panic(fmt.Sprintf("unknown transaction policy: %v", transactionPolicy))
// Any other task type is preserved in order.
// Eg: [START, UPSERT, TP1, CLOSE, TP2, TP3] -> [TP1, CLOSE, TP2, TP3]
func (ms *MutableStateImpl) closeTransactionCollapseVisibilityTasks() {
mutable_state_impl.go
visTasks := ms.InsertTasks[tasks.CategoryVisibility]
if len(visTasks) < 2 {
}
var visTaskToKeep tasks.Task
lastIndex := -1
}
func (ms *MutableStateImpl) generateChecksum() *persistencespb.Checksum {
mutable_state_impl.go
if !ms.shouldGenerateChecksum() {
return nil
}
csum, err := generateMutableStateChecksum(ms)
if err != nil {
}
if ms.namespaceEntry == nil {
return false
}
return rand.Intn(100) < ms.config.MutableStateChecksumGenProbability(ms.namespaceEntry.Name().String())
mutable_state_impl.go
}
//
// Note: Deployment objects are immutable, never change their fields.
func (ms *MutableStateImpl) GetEffectiveDeployment() *deploymentpb.Deployment {
mutable_state_impl.go
return GetEffectiveDeployment(ms.GetExecutionInfo().GetVersioningInfo())
}
func (ms *MutableStateImpl) GetWorkerDeploymentSA() string {
// 3. Behavior: this is returned when there is no override (most common case). Behavior is
// set based on the worker-sent deployment in the latest WFT completion.
func (ms *MutableStateImpl) GetEffectiveVersioningBehavior() enumspb.VersioningBehavior {
mutable_state_impl.go
return GetEffectiveVersioningBehavior(ms.GetExecutionInfo().GetVersioningInfo())
}
// StartDeploymentTransition starts a transition to the given deployment which must be
}
func (ms *MutableStateImpl) ToRealTime(virtualTime time.Time) time.Time {
mutable_state_impl.go
if virtualTime.IsZero() {
return virtualTime
}
}
dc *dynamicconfig.Collection,
numberOfShards int32,
cfg := &Config{
NumberOfShards: numberOfShards,
EnableReplicationStream: dynamicconfig.EnableReplicationStream.Get(dc),
EmitReplicationLifecycleEvents: dynamicconfig.EmitReplicationLifecycleEvents.Get(dc),
EnableCloseInboundReplicationStreamOnShutdown: dynamicconfig.EnableCloseInboundReplicationStreamOnShutdown.Get(dc),
EnableSeparateReplicationEnableFlag: dynamicconfig.EnableSeparateReplicationEnableFlag.Get(dc),
HistoryReplicationDLQV2: dynamicconfig.EnableHistoryReplicationDLQV2.Get(dc),
RPS: dynamicconfig.HistoryRPS.Get(dc),
NamespaceRPS: dynamicconfig.HistoryNamespaceRPS.Get(dc),
OperatorRPSRatio: dynamicconfig.OperatorRPSRatio.Get(dc),
MaxIDLengthLimit: dynamicconfig.MaxIDLengthLimit.Get(dc),
PersistenceMaxQPS: dynamicconfig.HistoryPersistenceMaxQPS.Get(dc),
PersistenceGlobalMaxQPS: dynamicconfig.HistoryPersistenceGlobalMaxQPS.Get(dc),
PersistenceNamespaceMaxQPS: dynamicconfig.HistoryPersistenceNamespaceMaxQPS.Get(dc),
PersistenceGlobalNamespaceMaxQPS: dynamicconfig.HistoryPersistenceGlobalNamespaceMaxQPS.Get(dc),
PersistencePerShardNamespaceMaxQPS: dynamicconfig.HistoryPersistencePerShardNamespaceMaxQPS.Get(dc),
PersistenceDynamicRateLimitingParams: dynamicconfig.HistoryPersistenceDynamicRateLimitingParams.Get(dc),
PersistenceQPSBurstRatio: dynamicconfig.PersistenceQPSBurstRatio.Get(dc),
AlignMembershipChange: dynamicconfig.HistoryAlignMembershipChange.Get(dc),
ShutdownDrainDuration: dynamicconfig.HistoryShutdownDrainDuration.Get(dc),
StartupMembershipJoinDelay: dynamicconfig.HistoryStartupMembershipJoinDelay.Get(dc),
AllowResetWithPendingChildren: dynamicconfig.AllowResetWithPendingChildren.Get(dc),
MaxAutoResetPoints: dynamicconfig.HistoryMaxAutoResetPoints.Get(dc),
DefaultWorkflowTaskTimeout: dynamicconfig.DefaultWorkflowTaskTimeout.Get(dc),
MaxLocalParentWorkflowVerificationDuration: dynamicconfig.MaxLocalParentWorkflowVerificationDuration.Get(dc),
VisibilityPersistenceMaxReadQPS: dynamicconfig.VisibilityPersistenceMaxReadQPS.Get(dc),
VisibilityPersistenceMaxWriteQPS: dynamicconfig.VisibilityPersistenceMaxWriteQPS.Get(dc),
VisibilityPersistenceSlowQueryThreshold: dynamicconfig.VisibilityPersistenceSlowQueryThreshold.Get(dc),
EnableReadFromSecondaryVisibility: dynamicconfig.EnableReadFromSecondaryVisibility.Get(dc),
VisibilityEnableShadowReadMode: dynamicconfig.VisibilityEnableShadowReadMode.Get(dc),
SecondaryVisibilityWritingMode: dynamicconfig.SecondaryVisibilityWritingMode.Get(dc),
VisibilityDisableOrderByClause: dynamicconfig.VisibilityDisableOrderByClause.Get(dc),
VisibilityEnableManualPagination: dynamicconfig.VisibilityEnableManualPagination.Get(dc),
VisibilityEnableUnifiedQueryConverter: dynamicconfig.VisibilityEnableUnifiedQueryConverter.Get(dc),
VisibilityAllowList: dynamicconfig.VisibilityAllowList.Get(dc),
SuppressErrorSetSystemSearchAttribute: dynamicconfig.SuppressErrorSetSystemSearchAttribute.Get(dc),
EmitShardLagLog: dynamicconfig.EmitShardLagLog.Get(dc),
EnableDataLossMetrics: dynamicconfig.EnableDataLossMetrics.Get(dc),
// HistoryCacheLimitSizeBased should not change during runtime.
HistoryCacheLimitSizeBased: dynamicconfig.HistoryCacheSizeBasedLimit.Get(dc)(),
HistoryHostLevelCacheMaxSize: dynamicconfig.HistoryCacheHostLevelMaxSize.Get(dc),
HistoryHostLevelCacheMaxSizeBytes: dynamicconfig.HistoryCacheHostLevelMaxSizeBytes.Get(dc),
HistoryCacheTTL: dynamicconfig.HistoryCacheTTL.Get(dc),
HistoryCacheNonUserContextLockTimeout: dynamicconfig.HistoryCacheNonUserContextLockTimeout.Get(dc),
HistoryCacheBackgroundEvict: dynamicconfig.HistoryCacheBackgroundEvict.Get(dc),
EnableWorkflowExecutionTimeoutTimer: dynamicconfig.EnableWorkflowExecutionTimeoutTimer.Get(dc),
EnableUpdateWorkflowModeIgnoreCurrent: dynamicconfig.EnableUpdateWorkflowModeIgnoreCurrent.Get(dc),
EnableTransitionHistory: dynamicconfig.EnableTransitionHistory.Get(dc),
MaxCallbacksPerWorkflow: dynamicconfig.MaxCallbacksPerWorkflow.Get(dc),
MaxCallbacksPerExecution: callback.MaxPerExecution.Get(dc),
MaxCallbacksPerUpdateID: dynamicconfig.MaxCallbacksPerUpdateID.Get(dc),
EnableChasm: dynamicconfig.EnableChasm.Get(dc),
EnableChasmNexusWorkflowOperations: nexusoperation.EnableChasmWorkflowOperations.Get(dc),
ChasmMaxInMemoryPureTasks: dynamicconfig.ChasmMaxInMemoryPureTasks.Get(dc),
EnableCHASMSchedulerCreation: dynamicconfig.EnableCHASMSchedulerCreation.Get(dc),
EnableCHASMSchedulerMigration: dynamicconfig.EnableCHASMSchedulerMigration.Get(dc),
EnableCHASMCallbacks: dynamicconfig.EnableCHASMCallbacks.Get(dc),
EnableCHASMSignalBacklinks: dynamicconfig.EnableCHASMSignalBacklinks.Get(dc),
ExternalPayloadsEnabled: dynamicconfig.ExternalPayloadsEnabled.Get(dc),
EnableWorkflowUpdateCallbacks: dynamicconfig.EnableWorkflowUpdateCallbacks.Get(dc),
EventsShardLevelCacheMaxSizeBytes: dynamicconfig.EventsCacheMaxSizeBytes.Get(dc), // 512KB
EventsHostLevelCacheMaxSizeBytes: dynamicconfig.EventsHostLevelCacheMaxSizeBytes.Get(dc), // 256MB
EventsCacheTTL: dynamicconfig.EventsCacheTTL.Get(dc),
EnableHostLevelEventsCache: dynamicconfig.EnableHostLevelEventsCache.Get(dc),
RangeSizeBits: 20, // 20 bits for sequencer, 2^20 sequence number for any range
AcquireShardInterval: dynamicconfig.AcquireShardInterval.Get(dc),
AcquireShardConcurrency: dynamicconfig.AcquireShardConcurrency.Get(dc),
ShardIOConcurrency: dynamicconfig.ShardIOConcurrency.Get(dc),
ShardIOTimeout: dynamicconfig.ShardIOTimeout.Get(dc),
ShardLingerOwnershipCheckQPS: dynamicconfig.ShardLingerOwnershipCheckQPS.Get(dc),
ShardLingerTimeLimit: dynamicconfig.ShardLingerTimeLimit.Get(dc),
ShardFinalizerTimeout: dynamicconfig.ShardFinalizerTimeout.Get(dc),
HistoryClientOwnershipCachingEnabled: dynamicconfig.HistoryClientOwnershipCachingEnabled.Get(dc),
StandbyClusterDelay: dynamicconfig.StandbyClusterDelay.Get(dc),
StandbyTaskMissingEventsResendDelay: dynamicconfig.StandbyTaskMissingEventsResendDelay.Get(dc),
StandbyTaskMissingEventsDiscardDelay: dynamicconfig.StandbyTaskMissingEventsDiscardDelay.Get(dc),
ChasmStandbyTaskDiscardDelay: dynamicconfig.ChasmStandbyTaskDiscardDelay.Get(dc),
QueuePendingTaskCriticalCount: dynamicconfig.QueuePendingTaskCriticalCount.Get(dc),
QueueReaderStuckCriticalAttempts: dynamicconfig.QueueReaderStuckCriticalAttempts.Get(dc),
QueueCriticalSlicesCount: dynamicconfig.QueueCriticalSlicesCount.Get(dc),
QueuePendingTaskMaxCount: dynamicconfig.QueuePendingTaskMaxCount.Get(dc),
QueueMaxPredicateSize: dynamicconfig.QueueMaxPredicateSize.Get(dc),
QueueShrinkPredicateMaxPendingKeys: dynamicconfig.QueueShrinkPredicateMaxPendingKeys.Get(dc),
QueueMoveGroupTaskCountBase: dynamicconfig.QueueMoveGroupTaskCountBase.Get(dc),
QueueMoveGroupTaskCountMultiplier: dynamicconfig.QueueMoveGroupTaskCountMultiplier.Get(dc),
TaskDLQEnabled: dynamicconfig.HistoryTaskDLQEnabled.Get(dc),
TaskDLQUnexpectedErrorAttempts: dynamicconfig.HistoryTaskDLQUnexpectedErrorAttempts.Get(dc),
TaskDLQInternalErrors: dynamicconfig.HistoryTaskDLQInternalErrors.Get(dc),
TaskDLQErrorPattern: dynamicconfig.HistoryTaskDLQErrorPattern.Get(dc),
TaskSchedulerEnableRateLimiter: dynamicconfig.TaskSchedulerEnableRateLimiter.Get(dc),
TaskSchedulerEnableRateLimiterShadowMode: dynamicconfig.TaskSchedulerEnableRateLimiterShadowMode.Get(dc),
TaskSchedulerRateLimiterStartupDelay: dynamicconfig.TaskSchedulerRateLimiterStartupDelay.Get(dc),
TaskSchedulerGlobalMaxQPS: dynamicconfig.TaskSchedulerGlobalMaxQPS.Get(dc),
TaskSchedulerMaxQPS: dynamicconfig.TaskSchedulerMaxQPS.Get(dc),
TaskSchedulerNamespaceMaxQPS: dynamicconfig.TaskSchedulerNamespaceMaxQPS.Get(dc),
TaskSchedulerGlobalNamespaceMaxQPS: dynamicconfig.TaskSchedulerGlobalNamespaceMaxQPS.Get(dc),
TaskSchedulerInactiveChannelDeletionDelay: dynamicconfig.TaskSchedulerInactiveChannelDeletionDelay.Get(dc),
TaskSchedulerEnableExecutionQueueScheduler: dynamicconfig.TaskSchedulerEnableExecutionQueueScheduler.Get(dc),
TaskSchedulerExecutionQueueSchedulerMaxQueues: dynamicconfig.TaskSchedulerExecutionQueueSchedulerMaxQueues.Get(dc),
TaskSchedulerExecutionQueueSchedulerQueueTTL: dynamicconfig.TaskSchedulerExecutionQueueSchedulerQueueTTL.Get(dc),
TaskSchedulerExecutionQueueSchedulerQueueConcurrency: dynamicconfig.TaskSchedulerExecutionQueueSchedulerQueueConcurrency.Get(dc),
TimerTaskBatchSize: dynamicconfig.TimerTaskBatchSize.Get(dc),
TimerProcessorSchedulerWorkerCount: dynamicconfig.TimerProcessorSchedulerWorkerCount.Subscribe(dc),
TimerProcessorSchedulerActiveRoundRobinWeights: dynamicconfig.TimerProcessorSchedulerActiveRoundRobinWeights.WithDefault(ConvertWeightsToDynamicConfigValue(DefaultActiveTaskPriorityWeight)).Get(dc),
TimerProcessorSchedulerStandbyRoundRobinWeights: dynamicconfig.TimerProcessorSchedulerStandbyRoundRobinWeights.WithDefault(ConvertWeightsToDynamicConfigValue(DefaultStandbyTaskPriorityWeight)).Get(dc),
TimerProcessorUpdateAckInterval: dynamicconfig.TimerProcessorUpdateAckInterval.Get(dc),
TimerProcessorUpdateAckIntervalJitterCoefficient: dynamicconfig.TimerProcessorUpdateAckIntervalJitterCoefficient.Get(dc),
TimerProcessorMaxPollRPS: dynamicconfig.TimerProcessorMaxPollRPS.Get(dc),
TimerProcessorMaxPollHostRPS: dynamicconfig.TimerProcessorMaxPollHostRPS.Get(dc),
TimerProcessorMaxPollInterval: dynamicconfig.TimerProcessorMaxPollInterval.Get(dc),
TimerProcessorMaxPollIntervalJitterCoefficient: dynamicconfig.TimerProcessorMaxPollIntervalJitterCoefficient.Get(dc),
TimerProcessorPollBackoffInterval: dynamicconfig.TimerProcessorPollBackoffInterval.Get(dc),
TimerProcessorMaxTimeShift: dynamicconfig.TimerProcessorMaxTimeShift.Get(dc),
TransferQueueMaxReaderCount: dynamicconfig.TransferQueueMaxReaderCount.Get(dc),
RetentionTimerJitterDuration: dynamicconfig.RetentionTimerJitterDuration.Get(dc),
MemoryTimerProcessorSchedulerWorkerCount: dynamicconfig.MemoryTimerProcessorSchedulerWorkerCount.Subscribe(dc),
TransferTaskBatchSize: dynamicconfig.TransferTaskBatchSize.Get(dc),
TransferProcessorSchedulerWorkerCount: dynamicconfig.TransferProcessorSchedulerWorkerCount.Subscribe(dc),
TransferProcessorSchedulerActiveRoundRobinWeights: dynamicconfig.TransferProcessorSchedulerActiveRoundRobinWeights.WithDefault(ConvertWeightsToDynamicConfigValue(DefaultActiveTaskPriorityWeight)).Get(dc),
TransferProcessorSchedulerStandbyRoundRobinWeights: dynamicconfig.TransferProcessorSchedulerStandbyRoundRobinWeights.WithDefault(ConvertWeightsToDynamicConfigValue(DefaultStandbyTaskPriorityWeight)).Get(dc),
TransferProcessorMaxPollRPS: dynamicconfig.TransferProcessorMaxPollRPS.Get(dc),
TransferProcessorMaxPollHostRPS: dynamicconfig.TransferProcessorMaxPollHostRPS.Get(dc),
TransferProcessorMaxPollInterval: dynamicconfig.TransferProcessorMaxPollInterval.Get(dc),
TransferProcessorMaxPollIntervalJitterCoefficient: dynamicconfig.TransferProcessorMaxPollIntervalJitterCoefficient.Get(dc),
TransferProcessorUpdateAckInterval: dynamicconfig.TransferProcessorUpdateAckInterval.Get(dc),
TransferProcessorUpdateAckIntervalJitterCoefficient: dynamicconfig.TransferProcessorUpdateAckIntervalJitterCoefficient.Get(dc),
TransferProcessorPollBackoffInterval: dynamicconfig.TransferProcessorPollBackoffInterval.Get(dc),
TransferProcessorEnsureCloseBeforeDelete: dynamicconfig.TransferProcessorEnsureCloseBeforeDelete.Get(dc),
TimerQueueMaxReaderCount: dynamicconfig.TimerQueueMaxReaderCount.Get(dc),
OutboundTaskBatchSize: dynamicconfig.OutboundTaskBatchSize.Get(dc),
OutboundProcessorMaxPollRPS: dynamicconfig.OutboundProcessorMaxPollRPS.Get(dc),
OutboundProcessorMaxPollHostRPS: dynamicconfig.OutboundProcessorMaxPollHostRPS.Get(dc),
OutboundProcessorMaxPollInterval: dynamicconfig.OutboundProcessorMaxPollInterval.Get(dc),
OutboundProcessorMaxPollIntervalJitterCoefficient: dynamicconfig.OutboundProcessorMaxPollIntervalJitterCoefficient.Get(dc),
OutboundProcessorUpdateAckInterval: dynamicconfig.OutboundProcessorUpdateAckInterval.Get(dc),
OutboundProcessorUpdateAckIntervalJitterCoefficient: dynamicconfig.OutboundProcessorUpdateAckIntervalJitterCoefficient.Get(dc),
OutboundProcessorPollBackoffInterval: dynamicconfig.OutboundProcessorPollBackoffInterval.Get(dc),
OutboundQueuePendingTaskCriticalCount: dynamicconfig.OutboundQueuePendingTaskCriticalCount.Get(dc),
OutboundQueuePendingTaskMaxCount: dynamicconfig.OutboundQueuePendingTaskMaxCount.Get(dc),
OutboundQueueMaxPredicateSize: dynamicconfig.OutboundQueueMaxPredicateSize.Get(dc),
OutboundQueueMaxReaderCount: dynamicconfig.OutboundQueueMaxReaderCount.Get(dc),
OutboundQueueGroupLimiterBufferSize: dynamicconfig.OutboundQueueGroupLimiterBufferSize.Get(dc),
OutboundQueueGroupLimiterConcurrency: dynamicconfig.OutboundQueueGroupLimiterConcurrency.Get(dc),
OutboundQueueHostSchedulerMaxTaskRPS: dynamicconfig.OutboundQueueHostSchedulerMaxTaskRPS.Get(dc),
OutboundQueueCircuitBreakerSettings: dynamicconfig.OutboundQueueCircuitBreakerSettings.Subscribe(dc),
OutboundStandbyTaskMissingEventsDestinationDownErr: dynamicconfig.OutboundStandbyTaskMissingEventsDestinationDownErr.Get(dc),
OutboundStandbyTaskMissingEventsDiscardDelay: dynamicconfig.OutboundStandbyTaskMissingEventsDiscardDelay.Get(dc),
ReplicatorProcessorMaxPollInterval: dynamicconfig.ReplicatorProcessorMaxPollInterval.Get(dc),
ReplicatorProcessorMaxPollIntervalJitterCoefficient: dynamicconfig.ReplicatorProcessorMaxPollIntervalJitterCoefficient.Get(dc),
ReplicatorProcessorFetchTasksBatchSize: dynamicconfig.ReplicatorTaskBatchSize.Get(dc),
ReplicatorProcessorMaxSkipTaskCount: dynamicconfig.ReplicatorMaxSkipTaskCount.Get(dc),
ReplicationTaskProcessorHostQPS: dynamicconfig.ReplicationTaskProcessorHostQPS.Get(dc),
ReplicationTaskProcessorShardQPS: dynamicconfig.ReplicationTaskProcessorShardQPS.Get(dc),
ReplicationEnableDLQMetrics: dynamicconfig.ReplicationEnableDLQMetrics.Get(dc),
ReplicationEnableUpdateWithNewTaskMerge: dynamicconfig.ReplicationEnableUpdateWithNewTaskMerge.Get(dc),
ReplicationStreamSyncStatusDuration: dynamicconfig.ReplicationStreamSyncStatusDuration.Get(dc),
ReplicationProcessorSchedulerQueueSize: dynamicconfig.ReplicationProcessorSchedulerQueueSize.Get(dc),
ReplicationProcessorSchedulerWorkerCount: dynamicconfig.ReplicationProcessorSchedulerWorkerCount.Subscribe(dc),
ReplicationLowPriorityProcessorSchedulerWorkerCount: dynamicconfig.ReplicationLowPriorityProcessorSchedulerWorkerCount.Subscribe(dc),
ReplicationLowPriorityTaskParallelism: dynamicconfig.ReplicationLowPriorityTaskParallelism.Get(dc),
EnableReplicationTaskBatching: dynamicconfig.EnableReplicationTaskBatching.Get(dc),
EnableReplicationTaskTieredProcessing: dynamicconfig.EnableReplicationTaskTieredProcessing.Get(dc),
ReplicationStreamSenderHighPriorityQPS: dynamicconfig.ReplicationStreamSenderHighPriorityQPS.Get(dc),
ReplicationStreamSenderLowPriorityQPS: dynamicconfig.ReplicationStreamSenderLowPriorityQPS.Get(dc),
ReplicationStreamEventLoopRetryMaxAttempts: dynamicconfig.ReplicationStreamEventLoopRetryMaxAttempts.Get(dc),
ReplicationReceiverMaxOutstandingTaskCount: dynamicconfig.ReplicationReceiverMaxOutstandingTaskCount.Get(dc),
ReplicationReceiverSlowSubmissionLatencyThreshold: dynamicconfig.ReplicationReceiverSlowSubmissionLatencyThreshold.Get(dc),
ReplicationReceiverSlowSubmissionWindow: dynamicconfig.ReplicationReceiverSlowSubmissionWindow.Get(dc),
EnableReplicationReceiverSlowSubmissionFlowControl: dynamicconfig.EnableReplicationReceiverSlowSubmissionFlowControl.Get(dc),
ReplicationResendMaxBatchCount: dynamicconfig.ReplicationResendMaxBatchCount.Get(dc),
ReplicationProgressCacheMaxSize: dynamicconfig.ReplicationProgressCacheMaxSize.Get(dc),
ReplicationProgressCacheTTL: dynamicconfig.ReplicationProgressCacheTTL.Get(dc),
ReplicationEnableRateLimit: dynamicconfig.ReplicationEnableRateLimit.Get(dc),
ReplicationEnableRateLimitShadowMode: dynamicconfig.ReplicationEnableRateLimitShadowMode.Get(dc),
ReplicationStreamSendEmptyTaskDuration: dynamicconfig.ReplicationStreamSendEmptyTaskDuration.Get(dc),
ReplicationStreamReceiverLivenessMultiplier: dynamicconfig.ReplicationStreamReceiverLivenessMultiplier.Get(dc),
ReplicationStreamSenderLivenessMultiplier: dynamicconfig.ReplicationStreamSenderLivenessMultiplier.Get(dc),
EnableHistoryReplicationRateLimiter: dynamicconfig.EnableHistoryReplicationRateLimiter.Get(dc),
MaximumBufferedEventsBatch: dynamicconfig.MaximumBufferedEventsBatch.Get(dc),
MaximumBufferedEventsSizeInBytes: dynamicconfig.MaximumBufferedEventsSizeInBytes.Get(dc),
MaximumSignalsPerExecution: dynamicconfig.MaximumSignalsPerExecution.Get(dc),
MaximumEventBatchSizeInBytes: dynamicconfig.MaximumEventBatchSizeInBytes.Get(dc),
ShardUpdateMinInterval: dynamicconfig.ShardUpdateMinInterval.Get(dc),
ShardFirstUpdateInterval: dynamicconfig.ShardFirstUpdateInterval.Get(dc),
ShardUpdateMinTasksCompleted: dynamicconfig.ShardUpdateMinTasksCompleted.Get(dc),
ShardSyncMinInterval: dynamicconfig.ShardSyncMinInterval.Get(dc),
ShardSyncTimerJitterCoefficient: dynamicconfig.TransferProcessorMaxPollIntervalJitterCoefficient.Get(dc),
// history client: client/history/client.go set the client timeout 30s
// TODO: Return this value to the client: go.temporal.io/server/issues/294
LongPollExpirationInterval: dynamicconfig.HistoryLongPollExpirationInterval.Get(dc),
EnableParentClosePolicy: dynamicconfig.EnableParentClosePolicy.Get(dc),
NumParentClosePolicySystemWorkflows: dynamicconfig.NumParentClosePolicySystemWorkflows.Get(dc),
EnableParentClosePolicyWorker: dynamicconfig.EnableParentClosePolicyWorker.Get(dc),
ParentClosePolicyThreshold: dynamicconfig.ParentClosePolicyThreshold.Get(dc),
BlobSizeLimitError: dynamicconfig.BlobSizeLimitError.Get(dc),
BlobSizeLimitWarn: dynamicconfig.BlobSizeLimitWarn.Get(dc),
MemoSizeLimitError: dynamicconfig.MemoSizeLimitError.Get(dc),
MemoSizeLimitWarn: dynamicconfig.MemoSizeLimitWarn.Get(dc),
NumPendingChildExecutionsLimit: dynamicconfig.NumPendingChildExecutionsLimitError.Get(dc),
NumPendingActivitiesLimit: dynamicconfig.NumPendingActivitiesLimitError.Get(dc),
NumPendingSignalsLimit: dynamicconfig.NumPendingSignalsLimitError.Get(dc),
NumPendingCancelsRequestLimit: dynamicconfig.NumPendingCancelRequestsLimitError.Get(dc),
HistorySizeLimitError: dynamicconfig.HistorySizeLimitError.Get(dc),
HistorySizeLimitWarn: dynamicconfig.HistorySizeLimitWarn.Get(dc),
HistorySizeSuggestContinueAsNew: dynamicconfig.HistorySizeSuggestContinueAsNew.Get(dc),
HistoryCountLimitError: dynamicconfig.HistoryCountLimitError.Get(dc),
HistoryCountLimitWarn: dynamicconfig.HistoryCountLimitWarn.Get(dc),
HistoryCountSuggestContinueAsNew: dynamicconfig.HistoryCountSuggestContinueAsNew.Get(dc),
HistoryMaxPageSize: dynamicconfig.HistoryMaxPageSize.Get(dc),
MutableStateActivityFailureSizeLimitError: dynamicconfig.MutableStateActivityFailureSizeLimitError.Get(dc),
MutableStateActivityFailureSizeLimitWarn: dynamicconfig.MutableStateActivityFailureSizeLimitWarn.Get(dc),
MutableStateSizeLimitError: dynamicconfig.MutableStateSizeLimitError.Get(dc),
MutableStateSizeLimitWarn: dynamicconfig.MutableStateSizeLimitWarn.Get(dc),
MutableStateTombstoneCountLimit: dynamicconfig.MutableStateTombstoneCountLimit.Get(dc),
ThrottledLogRPS: dynamicconfig.HistoryThrottledLogRPS.Get(dc),
EnableStickyQuery: dynamicconfig.EnableStickyQuery.Get(dc),
DefaultActivityRetryPolicy: dynamicconfig.DefaultActivityRetryPolicy.Get(dc),
DefaultWorkflowRetryPolicy: dynamicconfig.DefaultWorkflowRetryPolicy.Get(dc),
WorkflowTaskHeartbeatTimeout: dynamicconfig.WorkflowTaskHeartbeatTimeout.Get(dc),
WorkflowTaskCriticalAttempts: dynamicconfig.WorkflowTaskCriticalAttempts.Get(dc),
WorkflowTaskRetryMaxInterval: dynamicconfig.WorkflowTaskRetryMaxInterval.Get(dc),
EnableWorkflowTaskStampIncrementOnFailure: dynamicconfig.EnableWorkflowTaskStampIncrementOnFailure.Get(dc),
DiscardSpeculativeWorkflowTaskMaximumEventsCount: dynamicconfig.DiscardSpeculativeWorkflowTaskMaximumEventsCount.Get(dc),
EnableDropRepeatedWorkflowTaskFailures: dynamicconfig.EnableDropRepeatedWorkflowTaskFailures.Get(dc),
SendTransientOrSpeculativeWorkflowTaskEvents: dynamicconfig.SendTransientOrSpeculativeWorkflowTaskEvents.Get(dc),
ReplicationTaskApplyTimeout: dynamicconfig.ReplicationTaskApplyTimeout.Get(dc),
ReplicationTaskFetcherParallelism: dynamicconfig.ReplicationTaskFetcherParallelism.Get(dc),
ReplicationTaskFetcherAggregationInterval: dynamicconfig.ReplicationTaskFetcherAggregationInterval.Get(dc),
ReplicationTaskFetcherTimerJitterCoefficient: dynamicconfig.ReplicationTaskFetcherTimerJitterCoefficient.Get(dc),
ReplicationTaskFetcherErrorRetryWait: dynamicconfig.ReplicationTaskFetcherErrorRetryWait.Get(dc),
ReplicationTaskProcessorErrorRetryWait: dynamicconfig.ReplicationTaskProcessorErrorRetryWait.Get(dc),
ReplicationTaskProcessorErrorRetryBackoffCoefficient: dynamicconfig.ReplicationTaskProcessorErrorRetryBackoffCoefficient.Get(dc),
ReplicationTaskProcessorErrorRetryMaxInterval: dynamicconfig.ReplicationTaskProcessorErrorRetryMaxInterval.Get(dc),
ReplicationTaskProcessorErrorRetryMaxAttempts: dynamicconfig.ReplicationTaskProcessorErrorRetryMaxAttempts.Get(dc),
ReplicationTaskProcessorErrorRetryExpiration: dynamicconfig.ReplicationTaskProcessorErrorRetryExpiration.Get(dc),
ReplicationTaskProcessorNoTaskRetryWait: dynamicconfig.ReplicationTaskProcessorNoTaskInitialWait.Get(dc),
ReplicationTaskProcessorCleanupInterval: dynamicconfig.ReplicationTaskProcessorCleanupInterval.Get(dc),
ReplicationTaskProcessorCleanupJitterCoefficient: dynamicconfig.ReplicationTaskProcessorCleanupJitterCoefficient.Get(dc),
ReplicationMultipleBatches: dynamicconfig.ReplicationMultipleBatches.Get(dc),
ReplicationStreamSenderErrorRetryWait: dynamicconfig.ReplicationStreamSenderErrorRetryWait.Get(dc),
ReplicationStreamSenderErrorRetryBackoffCoefficient: dynamicconfig.ReplicationStreamSenderErrorRetryBackoffCoefficient.Get(dc),
ReplicationStreamSenderErrorRetryMaxInterval: dynamicconfig.ReplicationStreamSenderErrorRetryMaxInterval.Get(dc),
ReplicationStreamSenderErrorRetryMaxAttempts: dynamicconfig.ReplicationStreamSenderErrorRetryMaxAttempts.Get(dc),
ReplicationStreamSenderErrorRetryExpiration: dynamicconfig.ReplicationStreamSenderErrorRetryExpiration.Get(dc),
ReplicationExecutableTaskErrorRetryWait: dynamicconfig.ReplicationExecutableTaskErrorRetryWait.Get(dc),
ReplicationExecutableTaskErrorRetryBackoffCoefficient: dynamicconfig.ReplicationExecutableTaskErrorRetryBackoffCoefficient.Get(dc),
ReplicationExecutableTaskErrorRetryMaxInterval: dynamicconfig.ReplicationExecutableTaskErrorRetryMaxInterval.Get(dc),
ReplicationExecutableTaskErrorRetryMaxAttempts: dynamicconfig.ReplicationExecutableTaskErrorRetryMaxAttempts.Get(dc),
ReplicationExecutableTaskErrorRetryExpiration: dynamicconfig.ReplicationExecutableTaskErrorRetryExpiration.Get(dc),
MaxBufferedQueryCount: dynamicconfig.MaxBufferedQueryCount.Get(dc),
MutableStateChecksumGenProbability: dynamicconfig.MutableStateChecksumGenProbability.Get(dc),
MutableStateChecksumVerifyProbability: dynamicconfig.MutableStateChecksumVerifyProbability.Get(dc),
MutableStateChecksumInvalidateBefore: dynamicconfig.MutableStateChecksumInvalidateBefore.Get(dc),
StandbyTaskReReplicationContextTimeout: dynamicconfig.StandbyTaskReReplicationContextTimeout.Get(dc),
SkipReapplicationByNamespaceID: dynamicconfig.SkipReapplicationByNamespaceID.Get(dc),
// ===== Visibility related =====
VisibilityTaskBatchSize: dynamicconfig.VisibilityTaskBatchSize.Get(dc),
VisibilityProcessorMaxPollRPS: dynamicconfig.VisibilityProcessorMaxPollRPS.Get(dc),
VisibilityProcessorMaxPollHostRPS: dynamicconfig.VisibilityProcessorMaxPollHostRPS.Get(dc),
VisibilityProcessorSchedulerWorkerCount: dynamicconfig.VisibilityProcessorSchedulerWorkerCount.Subscribe(dc),
VisibilityProcessorSchedulerActiveRoundRobinWeights: dynamicconfig.VisibilityProcessorSchedulerActiveRoundRobinWeights.WithDefault(ConvertWeightsToDynamicConfigValue(DefaultActiveTaskPriorityWeight)).Get(dc),
VisibilityProcessorSchedulerStandbyRoundRobinWeights: dynamicconfig.VisibilityProcessorSchedulerStandbyRoundRobinWeights.WithDefault(ConvertWeightsToDynamicConfigValue(DefaultStandbyTaskPriorityWeight)).Get(dc),
VisibilityProcessorMaxPollInterval: dynamicconfig.VisibilityProcessorMaxPollInterval.Get(dc),
VisibilityProcessorMaxPollIntervalJitterCoefficient: dynamicconfig.VisibilityProcessorMaxPollIntervalJitterCoefficient.Get(dc),
VisibilityProcessorUpdateAckInterval: dynamicconfig.VisibilityProcessorUpdateAckInterval.Get(dc),
VisibilityProcessorUpdateAckIntervalJitterCoefficient: dynamicconfig.VisibilityProcessorUpdateAckIntervalJitterCoefficient.Get(dc),
VisibilityProcessorPollBackoffInterval: dynamicconfig.VisibilityProcessorPollBackoffInterval.Get(dc),
VisibilityProcessorEnsureCloseBeforeDelete: dynamicconfig.VisibilityProcessorEnsureCloseBeforeDelete.Get(dc),
VisibilityProcessorEnableCloseWorkflowCleanup: dynamicconfig.VisibilityProcessorEnableCloseWorkflowCleanup.Get(dc),
VisibilityProcessorRelocateAttributesMinBlobSize: dynamicconfig.VisibilityProcessorRelocateAttributesMinBlobSize.Get(dc),
VisibilityQueueMaxReaderCount: dynamicconfig.VisibilityQueueMaxReaderCount.Get(dc),
DisableFetchRelocatableAttributesFromVisibility: dynamicconfig.DisableFetchRelocatableAttributesFromVisibility.Get(dc),
SearchAttributesNumberOfKeysLimit: dynamicconfig.SearchAttributesNumberOfKeysLimit.Get(dc),
SearchAttributesSizeOfValueLimit: dynamicconfig.SearchAttributesSizeOfValueLimit.Get(dc),
SearchAttributesTotalSizeLimit: dynamicconfig.SearchAttributesTotalSizeLimit.Get(dc),
IndexerConcurrency: dynamicconfig.WorkerIndexerConcurrency.Get(dc),
ESProcessorNumOfWorkers: dynamicconfig.WorkerESProcessorNumOfWorkers.Get(dc),
// Should not be greater than number of visibility task queue workers VisibilityProcessorSchedulerWorkerCount (default 512)
// Otherwise, visibility queue processors won't be able to fill up bulk with documents (even under heavy load) and bulk will flush due to interval, not number of actions.
ESProcessorBulkActions: dynamicconfig.WorkerESProcessorBulkActions.Get(dc),
// 16MB - just a sanity check. With ES document size ~1Kb it should never be reached.
ESProcessorBulkSize: dynamicconfig.WorkerESProcessorBulkSize.Get(dc),
// Bulk processor will flush every this interval regardless of last flush due to bulk actions.
ESProcessorFlushInterval: dynamicconfig.WorkerESProcessorFlushInterval.Get(dc),
ESProcessorAckTimeout: dynamicconfig.WorkerESProcessorAckTimeout.Get(dc),
EnableCrossNamespaceCommands: dynamicconfig.EnableCrossNamespaceCommands.Get(dc),
EnableActivityEagerExecution: dynamicconfig.EnableActivityEagerExecution.Get(dc),
EnableActivityRetryStampIncrement: dynamicconfig.EnableActivityRetryStampIncrement.Get(dc),
EnableCancelActivityWorkerCommand: dynamicconfig.EnableCancelActivityWorkerCommand.Get(dc),
EnableEagerWorkflowStart: dynamicconfig.EnableEagerWorkflowStart.Get(dc),
NamespaceCacheRefreshInterval: dynamicconfig.NamespaceCacheRefreshInterval.Get(dc),
// Archival related
ArchivalTaskBatchSize: dynamicconfig.ArchivalTaskBatchSize.Get(dc),
ArchivalProcessorMaxPollRPS: dynamicconfig.ArchivalProcessorMaxPollRPS.Get(dc),
ArchivalProcessorMaxPollHostRPS: dynamicconfig.ArchivalProcessorMaxPollHostRPS.Get(dc),
ArchivalProcessorSchedulerWorkerCount: dynamicconfig.ArchivalProcessorSchedulerWorkerCount.Subscribe(dc),
ArchivalProcessorMaxPollInterval: dynamicconfig.ArchivalProcessorMaxPollInterval.Get(dc),
ArchivalProcessorMaxPollIntervalJitterCoefficient: dynamicconfig.ArchivalProcessorMaxPollIntervalJitterCoefficient.Get(dc),
ArchivalProcessorUpdateAckInterval: dynamicconfig.ArchivalProcessorUpdateAckInterval.Get(dc),
ArchivalProcessorUpdateAckIntervalJitterCoefficient: dynamicconfig.ArchivalProcessorUpdateAckIntervalJitterCoefficient.Get(dc),
ArchivalProcessorPollBackoffInterval: dynamicconfig.ArchivalProcessorPollBackoffInterval.Get(dc),
ArchivalProcessorArchiveDelay: dynamicconfig.ArchivalProcessorArchiveDelay.Get(dc),
ArchivalBackendMaxRPS: dynamicconfig.ArchivalBackendMaxRPS.Get(dc),
ArchivalQueueMaxReaderCount: dynamicconfig.ArchivalQueueMaxReaderCount.Get(dc),
// workflow update related
WorkflowExecutionMaxInFlightUpdates: dynamicconfig.WorkflowExecutionMaxInFlightUpdates.Get(dc),
WorkflowExecutionMaxInFlightUpdatePayloads: dynamicconfig.WorkflowExecutionMaxInFlightUpdatePayloads.Get(dc),
WorkflowExecutionMaxTotalUpdates: dynamicconfig.WorkflowExecutionMaxTotalUpdates.Get(dc),
WorkflowExecutionMaxTotalUpdatesSuggestContinueAsNewThreshold: dynamicconfig.WorkflowExecutionMaxTotalUpdatesSuggestContinueAsNewThreshold.Get(dc),
EnableUpdateWithStartRetryOnClosedWorkflowAbort: dynamicconfig.EnableUpdateWithStartRetryOnClosedWorkflowAbort.Get(dc),
EnableUpdateWithStartRetryableErrorOnClosedWorkflowAbort: dynamicconfig.EnableUpdateWithStartRetryableErrorOnClosedWorkflowAbort.Get(dc),
SendRawHistoryBetweenInternalServices: dynamicconfig.SendRawHistoryBetweenInternalServices.Get(dc),
SendRawHistoryBytesToMatchingService: dynamicconfig.SendRawHistoryBytesToMatchingService.Get(dc),
SendRawWorkflowHistory: dynamicconfig.SendRawWorkflowHistory.Get(dc),
WorkflowIdReuseMinimalInterval: dynamicconfig.WorkflowIdReuseMinimalInterval.Get(dc),
EnableWorkflowIdReuseStartTimeValidation: dynamicconfig.EnableWorkflowIdReuseStartTimeValidation.Get(dc),
BusinessIDReuseRate: dynamicconfig.BusinessIDReuseRate.Get(dc),
BusinessIDReuseBurstRatio: dynamicconfig.BusinessIDReuseBurstRatio.Get(dc),
BusinessIDReuseLimiterCacheSize: dynamicconfig.BusinessIDReuseLimiterCacheSize.Get(dc),
BusinessIDReuseLimiterCacheTTL: dynamicconfig.BusinessIDReuseLimiterCacheTTL.Get(dc),
HealthPersistenceLatencyFailure: dynamicconfig.HealthPersistenceLatencyFailure.Get(dc),
HealthPersistenceLatencyPercentiles: dynamicconfig.PersistenceHealthSignalPercentileLatencySettings.Get(dc),
HealthPersistenceErrorRatio: dynamicconfig.HealthPersistenceErrorRatio.Get(dc),
HealthRPCLatencyFailure: dynamicconfig.HealthRPCLatencyFailure.Get(dc),
HealthRPCLatencyPercentiles: dynamicconfig.HistoryHealthSignalPercentileLatencySettings.Get(dc),
HealthRPCErrorRatio: dynamicconfig.HealthRPCErrorRatio.Get(dc),
HealthHistoryInitializationTime: dynamicconfig.HealthHistoryInitializationTime.Get(dc),
BreakdownMetricsByTaskQueue: dynamicconfig.MetricsBreakdownByTaskQueue.Get(dc),
LogAllReqErrors: dynamicconfig.LogAllReqErrors.Get(dc),
NumConsecutiveWorkflowTaskProblemsToTriggerSearchAttribute: dynamicconfig.NumConsecutiveWorkflowTaskProblemsToTriggerSearchAttribute.Get(dc),
// Worker-Versioning related
UseRevisionNumberForWorkerVersioning: dynamicconfig.UseRevisionNumberForWorkerVersioning.Get(dc),
EnableSuggestCaNOnNewTargetVersion: dynamicconfig.EnableSuggestCaNOnNewTargetVersion.Get(dc),
EnableSendTargetVersionChanged: dynamicconfig.EnableSendTargetVersionChanged.Get(dc),
VersionMembershipCacheTTL: dynamicconfig.VersionMembershipCacheTTL.Get(dc),
VersionMembershipCacheMaxSize: dynamicconfig.VersionMembershipCacheMaxSize.Get(dc),
EnableVersionReactivationSignals: dynamicconfig.EnableVersionReactivationSignals.Get(dc),
RoutingInfoCacheTTL: dynamicconfig.RoutingInfoCacheTTL.Get(dc),
RoutingInfoCacheMaxSize: dynamicconfig.RoutingInfoCacheMaxSize.Get(dc),
// Workflow task completion pagination
EnableWorkflowTaskCompletionPagination: dynamicconfig.EnableWorkflowTaskCompletionPagination.Get(dc),
WorkflowTaskCompletionBufferSizeLimit: dynamicconfig.WorkflowTaskCompletionBufferSizeLimit.Get(dc),
}
return cfg
}
// GetShardID return the corresponding shard ID for a given namespaceID and workflowID pair
type GlobalBoolConstrainedDefaultSetting = GlobalTypedConstrainedDefaultSetting[bool]
func NewGlobalBoolSetting(key string, def bool, description string) GlobalBoolSetting {
setting_gen.go
return NewGlobalTypedSettingWithConverter[bool](key, convertBool, def, description)
}
func NewGlobalBoolSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[bool], description string) GlobalBoolConstrainedDefaultSetting {
type BoolPropertyFn = TypedPropertyFn[bool]
return GetTypedPropertyFn(value)
}
type NamespaceBoolSetting = NamespaceTypedSetting[bool]
type NamespaceBoolConstrainedDefaultSetting = NamespaceTypedConstrainedDefaultSetting[bool]
func NewNamespaceBoolSetting(key string, def bool, description string) NamespaceBoolSetting {
setting_gen.go
return NewNamespaceTypedSettingWithConverter[bool](key, convertBool, def, description)
}
func NewNamespaceBoolSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[bool], description string) NamespaceBoolConstrainedDefaultSetting {
type BoolPropertyFnWithNamespaceFilter = TypedPropertyFnWithNamespaceFilter[bool]
func GetBoolPropertyFnFilteredByNamespace(value bool) BoolPropertyFnWithNamespaceFilter {
setting_gen.go
return GetTypedPropertyFnFilteredByNamespace(value)
}
type NamespaceIDBoolSetting = NamespaceIDTypedSetting[bool]
type NamespaceIDBoolConstrainedDefaultSetting = NamespaceIDTypedConstrainedDefaultSetting[bool]
func NewNamespaceIDBoolSetting(key string, def bool, description string) NamespaceIDBoolSetting {
setting_gen.go
return NewNamespaceIDTypedSettingWithConverter[bool](key, convertBool, def, description)
}
func NewNamespaceIDBoolSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[bool], description string) NamespaceIDBoolConstrainedDefaultSetting {
type TaskQueueBoolConstrainedDefaultSetting = TaskQueueTypedConstrainedDefaultSetting[bool]
func NewTaskQueueBoolSetting(key string, def bool, description string) TaskQueueBoolSetting {
setting_gen.go
return NewTaskQueueTypedSettingWithConverter[bool](key, convertBool, def, description)
}
func NewTaskQueueBoolSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[bool], description string) TaskQueueBoolConstrainedDefaultSetting {
type DestinationBoolConstrainedDefaultSetting = DestinationTypedConstrainedDefaultSetting[bool]
func NewDestinationBoolSetting(key string, def bool, description string) DestinationBoolSetting {
setting_gen.go
return NewDestinationTypedSettingWithConverter[bool](key, convertBool, def, description)
}
func NewDestinationBoolSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[bool], description string) DestinationBoolConstrainedDefaultSetting {
type GlobalIntConstrainedDefaultSetting = GlobalTypedConstrainedDefaultSetting[int]
func NewGlobalIntSetting(key string, def int, description string) GlobalIntSetting {
setting_gen.go
return NewGlobalTypedSettingWithConverter[int](key, convertInt, def, description)
}
func NewGlobalIntSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[int], description string) GlobalIntConstrainedDefaultSetting {
type NamespaceIntConstrainedDefaultSetting = NamespaceTypedConstrainedDefaultSetting[int]
func NewNamespaceIntSetting(key string, def int, description string) NamespaceIntSetting {
setting_gen.go
return NewNamespaceTypedSettingWithConverter[int](key, convertInt, def, description)
}
func NewNamespaceIntSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[int], description string) NamespaceIntConstrainedDefaultSetting {
type IntPropertyFnWithNamespaceFilter = TypedPropertyFnWithNamespaceFilter[int]
func GetIntPropertyFnFilteredByNamespace(value int) IntPropertyFnWithNamespaceFilter {
setting_gen.go
return GetTypedPropertyFnFilteredByNamespace(value)
}
type NamespaceIDIntSetting = NamespaceIDTypedSetting[int]
type TaskQueueIntConstrainedDefaultSetting = TaskQueueTypedConstrainedDefaultSetting[int]
func NewTaskQueueIntSetting(key string, def int, description string) TaskQueueIntSetting {
setting_gen.go
return NewTaskQueueTypedSettingWithConverter[int](key, convertInt, def, description)
}
func NewTaskQueueIntSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[int], description string) TaskQueueIntConstrainedDefaultSetting {
setting_gen.go
return NewTaskQueueTypedSettingWithConstrainedDefault[int](key, convertInt, cdef, description)
}
type IntPropertyFnWithTaskQueueFilter = TypedPropertyFnWithTaskQueueFilter[int]
type ShardIDIntConstrainedDefaultSetting = ShardIDTypedConstrainedDefaultSetting[int]
func NewShardIDIntSetting(key string, def int, description string) ShardIDIntSetting {
setting_gen.go
return NewShardIDTypedSettingWithConverter[int](key, convertInt, def, description)
}
func NewShardIDIntSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[int], description string) ShardIDIntConstrainedDefaultSetting {
type DestinationIntConstrainedDefaultSetting = DestinationTypedConstrainedDefaultSetting[int]
func NewDestinationIntSetting(key string, def int, description string) DestinationIntSetting {
setting_gen.go
return NewDestinationTypedSettingWithConverter[int](key, convertInt, def, description)
}
func NewDestinationIntSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[int], description string) DestinationIntConstrainedDefaultSetting {
type GlobalFloatConstrainedDefaultSetting = GlobalTypedConstrainedDefaultSetting[float64]
func NewGlobalFloatSetting(key string, def float64, description string) GlobalFloatSetting {
setting_gen.go
return NewGlobalTypedSettingWithConverter[float64](key, convertFloat, def, description)
}
func NewGlobalFloatSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[float64], description string) GlobalFloatConstrainedDefaultSetting {
type NamespaceFloatConstrainedDefaultSetting = NamespaceTypedConstrainedDefaultSetting[float64]
func NewNamespaceFloatSetting(key string, def float64, description string) NamespaceFloatSetting {
setting_gen.go
return NewNamespaceTypedSettingWithConverter[float64](key, convertFloat, def, description)
}
func NewNamespaceFloatSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[float64], description string) NamespaceFloatConstrainedDefaultSetting {
type TaskQueueFloatConstrainedDefaultSetting = TaskQueueTypedConstrainedDefaultSetting[float64]
func NewTaskQueueFloatSetting(key string, def float64, description string) TaskQueueFloatSetting {
setting_gen.go
return NewTaskQueueTypedSettingWithConverter[float64](key, convertFloat, def, description)
}
func NewTaskQueueFloatSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[float64], description string) TaskQueueFloatConstrainedDefaultSetting {
type ShardIDFloatConstrainedDefaultSetting = ShardIDTypedConstrainedDefaultSetting[float64]
func NewShardIDFloatSetting(key string, def float64, description string) ShardIDFloatSetting {
setting_gen.go
return NewShardIDTypedSettingWithConverter[float64](key, convertFloat, def, description)
}
func NewShardIDFloatSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[float64], description string) ShardIDFloatConstrainedDefaultSetting {
type DestinationFloatConstrainedDefaultSetting = DestinationTypedConstrainedDefaultSetting[float64]
func NewDestinationFloatSetting(key string, def float64, description string) DestinationFloatSetting {
setting_gen.go
return NewDestinationTypedSettingWithConverter[float64](key, convertFloat, def, description)
}
func NewDestinationFloatSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[float64], description string) DestinationFloatConstrainedDefaultSetting {
type GlobalStringConstrainedDefaultSetting = GlobalTypedConstrainedDefaultSetting[string]
func NewGlobalStringSetting(key string, def string, description string) GlobalStringSetting {
setting_gen.go
return NewGlobalTypedSettingWithConverter[string](key, convertString, def, description)
}
func NewGlobalStringSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[string], description string) GlobalStringConstrainedDefaultSetting {
type GlobalDurationConstrainedDefaultSetting = GlobalTypedConstrainedDefaultSetting[time.Duration]
func NewGlobalDurationSetting(key string, def time.Duration, description string) GlobalDurationSetting {
setting_gen.go
return NewGlobalTypedSettingWithConverter[time.Duration](key, convertDuration, def, description)
}
func NewGlobalDurationSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[time.Duration], description string) GlobalDurationConstrainedDefaultSetting {
type DurationPropertyFn = TypedPropertyFn[time.Duration]
return GetTypedPropertyFn(value)
}
type NamespaceDurationSetting = NamespaceTypedSetting[time.Duration]
type NamespaceDurationConstrainedDefaultSetting = NamespaceTypedConstrainedDefaultSetting[time.Duration]
func NewNamespaceDurationSetting(key string, def time.Duration, description string) NamespaceDurationSetting {
setting_gen.go
return NewNamespaceTypedSettingWithConverter[time.Duration](key, convertDuration, def, description)
}
func NewNamespaceDurationSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[time.Duration], description string) NamespaceDurationConstrainedDefaultSetting {
type NamespaceIDDurationConstrainedDefaultSetting = NamespaceIDTypedConstrainedDefaultSetting[time.Duration]
func NewNamespaceIDDurationSetting(key string, def time.Duration, description string) NamespaceIDDurationSetting {
setting_gen.go
return NewNamespaceIDTypedSettingWithConverter[time.Duration](key, convertDuration, def, description)
}
func NewNamespaceIDDurationSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[time.Duration], description string) NamespaceIDDurationConstrainedDefaultSetting {
type TaskQueueDurationConstrainedDefaultSetting = TaskQueueTypedConstrainedDefaultSetting[time.Duration]
func NewTaskQueueDurationSetting(key string, def time.Duration, description string) TaskQueueDurationSetting {
setting_gen.go
return NewTaskQueueTypedSettingWithConverter[time.Duration](key, convertDuration, def, description)
}
func NewTaskQueueDurationSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[time.Duration], description string) TaskQueueDurationConstrainedDefaultSetting {
setting_gen.go
return NewTaskQueueTypedSettingWithConstrainedDefault[time.Duration](key, convertDuration, cdef, description)
}
type DurationPropertyFnWithTaskQueueFilter = TypedPropertyFnWithTaskQueueFilter[time.Duration]
type ShardIDDurationConstrainedDefaultSetting = ShardIDTypedConstrainedDefaultSetting[time.Duration]
func NewShardIDDurationSetting(key string, def time.Duration, description string) ShardIDDurationSetting {
setting_gen.go
return NewShardIDTypedSettingWithConverter[time.Duration](key, convertDuration, def, description)
}
func NewShardIDDurationSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[time.Duration], description string) ShardIDDurationConstrainedDefaultSetting {
type TaskTypeDurationConstrainedDefaultSetting = TaskTypeTypedConstrainedDefaultSetting[time.Duration]
func NewTaskTypeDurationSetting(key string, def time.Duration, description string) TaskTypeDurationSetting {
setting_gen.go
return NewTaskTypeTypedSettingWithConverter[time.Duration](key, convertDuration, def, description)
}
func NewTaskTypeDurationSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[time.Duration], description string) TaskTypeDurationConstrainedDefaultSetting {
type DestinationDurationConstrainedDefaultSetting = DestinationTypedConstrainedDefaultSetting[time.Duration]
func NewDestinationDurationSetting(key string, def time.Duration, description string) DestinationDurationSetting {
setting_gen.go
return NewDestinationTypedSettingWithConverter[time.Duration](key, convertDuration, def, description)
}
func NewDestinationDurationSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[time.Duration], description string) DestinationDurationConstrainedDefaultSetting {
type ChasmTaskTypeDurationConstrainedDefaultSetting = ChasmTaskTypeTypedConstrainedDefaultSetting[time.Duration]
func NewChasmTaskTypeDurationSetting(key string, def time.Duration, description string) ChasmTaskTypeDurationSetting {
setting_gen.go
return NewChasmTaskTypeTypedSettingWithConverter[time.Duration](key, convertDuration, def, description)
}
func NewChasmTaskTypeDurationSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[time.Duration], description string) ChasmTaskTypeDurationConstrainedDefaultSetting {
type NamespaceMapConstrainedDefaultSetting = NamespaceTypedConstrainedDefaultSetting[map[string]any]
func NewNamespaceMapSetting(key string, def map[string]any, description string) NamespaceMapSetting {
setting_gen.go
return NewNamespaceTypedSettingWithConverter[map[string]any](key, convertMap, def, description)
}
func NewNamespaceMapSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[map[string]any], description string) NamespaceMapConstrainedDefaultSetting {
// values. The value from dynamic config will be _merged_ over a deep copy of 'def'. Be very careful
// when using non-empty maps or slices as defaults, the result may not be what you want.
func NewGlobalTypedSetting[T any](key string, def T, description string) GlobalTypedSetting[T] {
setting_gen.go
// Warn on any shared structure used with ConvertStructure, even though we handle it by deep copying.
warnDefaultSharedStructure(key, def)
// If even deep copy won't even work, we should panic early. Do that by calling deep copy once here.
_ = deepCopyForMapstructure(def)
s := GlobalTypedSetting[T]{
key: MakeKey(key),
def: def,
convert: ConvertStructure[T](def),
description: description,
}
register(s)
return s
}
// NewGlobalTypedSettingWithConverter creates a setting with a custom converter function.
func NewGlobalTypedSettingWithConverter[T any](key string, convert func(any) (T, error), def T, description string) GlobalTypedSetting[T] {
setting_gen.go
s := GlobalTypedSetting[T]{
key: MakeKey(key),
def: def,
convert: convert,
description: description,
}
register(s)
return s
}
// NewGlobalTypedSettingWithConstrainedDefault creates a setting with a compound default value.
}
func (s GlobalTypedSetting[T]) Precedence() Precedence { return PrecedenceGlobal }
func (s GlobalTypedSetting[T]) Validate(v any) error {
type TypedPropertyFn[T any] func() T
return func() T {
return matchAndConvert(
c,
s.key,
s.def,
s.convert,
prec,
)
}
}
type TypedSubscribable[T any] func(callback func(T)) (v T, cancel func())
return func(callback func(T)) (T, func()) {
prec := []Constraints{{}}
return subscribe(c, s.key, s.def, s.convert, prec, callback)
}
return func() T {
return value
}
// values. The value from dynamic config will be _merged_ over a deep copy of 'def'. Be very careful
// when using non-empty maps or slices as defaults, the result may not be what you want.
func NewNamespaceTypedSetting[T any](key string, def T, description string) NamespaceTypedSetting[T] {
setting_gen.go
// Warn on any shared structure used with ConvertStructure, even though we handle it by deep copying.
warnDefaultSharedStructure(key, def)
// If even deep copy won't even work, we should panic early. Do that by calling deep copy once here.
_ = deepCopyForMapstructure(def)
s := NamespaceTypedSetting[T]{
key: MakeKey(key),
def: def,
convert: ConvertStructure[T](def),
description: description,
}
register(s)
return s
}
// NewNamespaceTypedSettingWithConverter creates a setting with a custom converter function.
func NewNamespaceTypedSettingWithConverter[T any](key string, convert func(any) (T, error), def T, description string) NamespaceTypedSetting[T] {
setting_gen.go
s := NamespaceTypedSetting[T]{
key: MakeKey(key),
def: def,
convert: convert,
description: description,
}
register(s)
return s
}
// NewNamespaceTypedSettingWithConstrainedDefault creates a setting with a compound default value.
}
func (s NamespaceTypedSetting[T]) Precedence() Precedence { return PrecedenceNamespace }
func (s NamespaceTypedSetting[T]) Validate(v any) error {
}
newS := s
newS.def = v
return newS
}
type TypedPropertyFnWithNamespaceFilter[T any] func(namespace string) T
func (s NamespaceTypedSetting[T]) Get(c *Collection) TypedPropertyFnWithNamespaceFilter[T] {
setting_gen.go
return func(namespace string) T {
return matchAndConvert(
c,
s.key,
s.def,
s.convert,
prec,
)
}
}
}
func GetTypedPropertyFnFilteredByNamespace[T any](value T) TypedPropertyFnWithNamespaceFilter[T] {
setting_gen.go
return func(namespace string) T {
}
}
// NewNamespaceIDTypedSettingWithConverter creates a setting with a custom converter function.
func NewNamespaceIDTypedSettingWithConverter[T any](key string, convert func(any) (T, error), def T, description string) NamespaceIDTypedSetting[T] {
setting_gen.go
s := NamespaceIDTypedSetting[T]{
key: MakeKey(key),
def: def,
convert: convert,
description: description,
}
register(s)
return s
}
// NewNamespaceIDTypedSettingWithConstrainedDefault creates a setting with a compound default value.
}
func (s NamespaceIDTypedSetting[T]) Precedence() Precedence { return PrecedenceNamespaceID }
func (s NamespaceIDTypedSetting[T]) Validate(v any) error {
type TypedPropertyFnWithNamespaceIDFilter[T any] func(namespaceID namespace.ID) T
func (s NamespaceIDTypedSetting[T]) Get(c *Collection) TypedPropertyFnWithNamespaceIDFilter[T] {
setting_gen.go
return func(namespaceID namespace.ID) T {
prec := []Constraints{{NamespaceID: namespaceID.String()}, {}}
return matchAndConvert(
// values. The value from dynamic config will be _merged_ over a deep copy of 'def'. Be very careful
// when using non-empty maps or slices as defaults, the result may not be what you want.
func NewTaskQueueTypedSetting[T any](key string, def T, description string) TaskQueueTypedSetting[T] {
setting_gen.go
// Warn on any shared structure used with ConvertStructure, even though we handle it by deep copying.
warnDefaultSharedStructure(key, def)
// If even deep copy won't even work, we should panic early. Do that by calling deep copy once here.
_ = deepCopyForMapstructure(def)
s := TaskQueueTypedSetting[T]{
key: MakeKey(key),
def: def,
convert: ConvertStructure[T](def),
description: description,
}
register(s)
return s
}
// NewTaskQueueTypedSettingWithConverter creates a setting with a custom converter function.
func NewTaskQueueTypedSettingWithConverter[T any](key string, convert func(any) (T, error), def T, description string) TaskQueueTypedSetting[T] {
setting_gen.go
s := TaskQueueTypedSetting[T]{
key: MakeKey(key),
def: def,
convert: convert,
description: description,
}
register(s)
return s
}
// NewTaskQueueTypedSettingWithConstrainedDefault creates a setting with a compound default value.
func NewTaskQueueTypedSettingWithConstrainedDefault[T any](key string, convert func(any) (T, error), cdef []TypedConstrainedValue[T], description string) TaskQueueTypedConstrainedDefaultSetting[T] {
setting_gen.go
s := TaskQueueTypedConstrainedDefaultSetting[T]{
key: MakeKey(key),
cdef: cdef,
convert: convert,
description: description,
}
register(s)
return s
}
func (s TaskQueueTypedSetting[T]) Precedence() Precedence { return PrecedenceTaskQueue }
func (s TaskQueueTypedSetting[T]) Validate(v any) error {
}
func (s TaskQueueTypedConstrainedDefaultSetting[T]) Key() Key { return s.key }
setting_gen.go
func (s TaskQueueTypedConstrainedDefaultSetting[T]) Precedence() Precedence { return PrecedenceTaskQueue }
func (s TaskQueueTypedConstrainedDefaultSetting[T]) Validate(v any) error {
type TypedPropertyFnWithTaskQueueFilter[T any] func(namespace string, taskQueue string, taskQueueType enumspb.TaskQueueType) T
func (s TaskQueueTypedSetting[T]) Get(c *Collection) TypedPropertyFnWithTaskQueueFilter[T] {
setting_gen.go
return func(namespace string, taskQueue string, taskQueueType enumspb.TaskQueueType) T {
prec := []Constraints{
{Namespace: namespace, TaskQueueName: taskQueue, TaskQueueType: taskQueueType},
// NewShardIDTypedSettingWithConverter creates a setting with a custom converter function.
func NewShardIDTypedSettingWithConverter[T any](key string, convert func(any) (T, error), def T, description string) ShardIDTypedSetting[T] {
setting_gen.go
s := ShardIDTypedSetting[T]{
key: MakeKey(key),
def: def,
convert: convert,
description: description,
}
register(s)
return s
}
// NewShardIDTypedSettingWithConstrainedDefault creates a setting with a compound default value.
}
func (s ShardIDTypedSetting[T]) Precedence() Precedence { return PrecedenceShardID }
func (s ShardIDTypedSetting[T]) Validate(v any) error {
type TypedPropertyFnWithShardIDFilter[T any] func(shardID int32) T
func (s ShardIDTypedSetting[T]) Get(c *Collection) TypedPropertyFnWithShardIDFilter[T] {
setting_gen.go
return func(shardID int32) T {
prec := []Constraints{{ShardID: shardID}, {}}
return matchAndConvert(
// NewTaskTypeTypedSettingWithConverter creates a setting with a custom converter function.
func NewTaskTypeTypedSettingWithConverter[T any](key string, convert func(any) (T, error), def T, description string) TaskTypeTypedSetting[T] {
setting_gen.go
s := TaskTypeTypedSetting[T]{
key: MakeKey(key),
def: def,
convert: convert,
description: description,
}
register(s)
return s
}
// NewTaskTypeTypedSettingWithConstrainedDefault creates a setting with a compound default value.
}
func (s TaskTypeTypedSetting[T]) Precedence() Precedence { return PrecedenceTaskType }
func (s TaskTypeTypedSetting[T]) Validate(v any) error {
type TypedPropertyFnWithTaskTypeFilter[T any] func(taskType enumsspb.TaskType) T
func (s TaskTypeTypedSetting[T]) Get(c *Collection) TypedPropertyFnWithTaskTypeFilter[T] {
setting_gen.go
return func(taskType enumsspb.TaskType) T {
prec := []Constraints{{TaskType: taskType}, {}}
return matchAndConvert(
// values. The value from dynamic config will be _merged_ over a deep copy of 'def'. Be very careful
// when using non-empty maps or slices as defaults, the result may not be what you want.
func NewDestinationTypedSetting[T any](key string, def T, description string) DestinationTypedSetting[T] {
setting_gen.go
// Warn on any shared structure used with ConvertStructure, even though we handle it by deep copying.
warnDefaultSharedStructure(key, def)
// If even deep copy won't even work, we should panic early. Do that by calling deep copy once here.
_ = deepCopyForMapstructure(def)
s := DestinationTypedSetting[T]{
key: MakeKey(key),
def: def,
convert: ConvertStructure[T](def),
description: description,
}
register(s)
return s
}
// NewDestinationTypedSettingWithConverter creates a setting with a custom converter function.
func NewDestinationTypedSettingWithConverter[T any](key string, convert func(any) (T, error), def T, description string) DestinationTypedSetting[T] {
setting_gen.go
s := DestinationTypedSetting[T]{
key: MakeKey(key),
def: def,
convert: convert,
description: description,
}
register(s)
return s
}
// NewDestinationTypedSettingWithConstrainedDefault creates a setting with a compound default value.
}
func (s DestinationTypedSetting[T]) Precedence() Precedence { return PrecedenceDestination }
func (s DestinationTypedSetting[T]) Validate(v any) error {
type TypedPropertyFnWithDestinationFilter[T any] func(namespace string, destination string) T
func (s DestinationTypedSetting[T]) Get(c *Collection) TypedPropertyFnWithDestinationFilter[T] {
setting_gen.go
return func(namespace string, destination string) T {
prec := []Constraints{
{Namespace: namespace, Destination: destination},
type TypedSubscribableWithDestinationFilter[T any] func(namespace string, destination string, callback func(T)) (v T, cancel func())
func (s DestinationTypedSetting[T]) Subscribe(c *Collection) TypedSubscribableWithDestinationFilter[T] {
setting_gen.go
return func(namespace string, destination string, callback func(T)) (T, func()) {
prec := []Constraints{
{Namespace: namespace, Destination: destination},
// NewChasmTaskTypeTypedSettingWithConverter creates a setting with a custom converter function.
func NewChasmTaskTypeTypedSettingWithConverter[T any](key string, convert func(any) (T, error), def T, description string) ChasmTaskTypeTypedSetting[T] {
setting_gen.go
s := ChasmTaskTypeTypedSetting[T]{
key: MakeKey(key),
def: def,
convert: convert,
description: description,
}
register(s)
return s
}
// NewChasmTaskTypeTypedSettingWithConstrainedDefault creates a setting with a compound default value.
}
func (s ChasmTaskTypeTypedSetting[T]) Precedence() Precedence { return PrecedenceChasmTaskType }
func (s ChasmTaskTypeTypedSetting[T]) Validate(v any) error {
type TypedPropertyFnWithChasmTaskTypeFilter[T any] func(chasmTaskType string) T
func (s ChasmTaskTypeTypedSetting[T]) Get(c *Collection) TypedPropertyFnWithChasmTaskTypeFilter[T] {
setting_gen.go
return func(chasmTaskType string) T {
prec := []Constraints{{ChasmTaskType: chasmTaskType}, {}}
return matchAndConvert(
}
// constant from initialization, no need for locks
return s.shardID
}
func (s *ContextImpl) GetRangeID() int64 {
}
// constant from initialization, no need for locks
return s.executionManager
}
func (s *ContextImpl) GetPingChecks() []pingable.Check {
}
s.rLock()
defer s.rUnlock()
nextTaskKey := s.taskKeyManager.peekTaskKey(tasks.CategoryTransfer)
return vclock.NewVectorClock(s.clusterMetadata.GetClusterID(), s.shardID, nextTaskKey.TaskID)
}
func (s *ContextImpl) GenerateTaskID() (int64, error) {
}
s.wLock()
defer s.wUnlock()
result := []int64{}
for range number {
if err != nil {
return nil, err
}
}
}
}
// constant from initialization, no need for locks
return s.config
}
// constant from initialization (except for tests), no need for locks
return s.eventsCache
}
// constant from initialization, no need for locks
return s.contextTaggedLogger
}
func (s *ContextImpl) GetThrottledLogger() log.Logger {
}
taskKey, err := s.taskKeyManager.generateTaskKey(tasks.CategoryTransfer)
if err != nil {
return -1, err
}
}
// FinishStop should only be called by the controller.
// After this returns, engineFuture.Set may not be called anymore, so if we don't get see
// an Engine here, we won't ever have one.
_ = s.transition(contextRequestFinishStop{})
// Use a context that we know is cancelled so that this doesn't block.
engine, _ := s.engineFuture.Get(s.lifecycleCtx)
// Stop the engine if it was running (outside the lock but before returning).
if engine != nil {
s.contextTaggedLogger.Info("", tag.LifeCycleStopping, tag.ComponentShardEngine)
engine.Stop()
// Run finalizer to cleanup any of the shard's associated resources that are registered.
s.finalizer.Run(s.config.ShardFinalizerTimeout())
}
}
handler := s.metricsHandler.WithTags(metrics.OperationTag(metrics.ShardInfoScope))
metrics.LockRequests.With(handler).Record(1)
startTime := time.Now().UTC()
defer func() { metrics.LockLatency.With(handler).Record(time.Since(startTime)) }()
}
handler := s.metricsHandler.WithTags(metrics.OperationTag(metrics.ShardInfoScope))
metrics.LockRequests.With(handler).Record(1)
startTime := time.Now().UTC()
defer func() { metrics.LockLatency.With(handler).Record(time.Since(startTime)) }()
}
s.rwLock.Unlock()
}
s.rwLock.RUnlock()
}
func (s *ContextImpl) ioSemaphoreAcquire(
}
/* State transitions:
The normal pattern:
Initialized
controller calls start()
Acquiring
acquireShard gets the shard
Acquired
If we get a transient error from persistence:
Acquired
transient error: handleErrorLocked calls transition(contextRequestLost)
Acquiring
acquireShard gets the shard
Acquired
If we get shard ownership lost:
Acquired
ShardOwnershipLostError: handleErrorLocked calls transition(contextRequestStop)
Stopping
controller removes from map and calls FinishStop()
Stopped
Stopping can be triggered internally (if we get a ShardOwnershipLostError, or fail to acquire the rangeid
lock after several minutes) or externally (from controller, e.g. controller shutting down or admin force-
unload shard). If it's triggered internally, we transition to Stopping, then make an asynchronous callback
to controller, which will remove us from the map and call FinishStop(), which will transition to Stopped and
stop the engine. If it's triggered externally, we'll skip over Stopping and go straight to Stopped.
If we transition externally to Stopped, and the acquireShard goroutine is still running, we can't kill it,
but we should make sure that it can't do anything: the context it uses for persistence ops will be
canceled, and if it tries to transition states, it will fail.
Invariants:
- Once state is Stopping, it can only go to Stopped.
- Once state is Stopped, it can't go anywhere else.
- At the start of acquireShard, state must be Acquiring.
- By the end of acquireShard, state must not be Acquiring: either acquireShard set it to Acquired, or the
controller set it to Stopped.
- If state is Acquiring, acquireShard should be running in the background.
- Only acquireShard can use contextRequestAcquired (i.e. transition from Acquiring to Acquired).
- Once state has reached Acquired at least once, and not reached Stopped, engineFuture must be set.
- Only the controller may call start() and FinishStop().
- The controller must call FinishStop() for every ContextImpl it creates.
*/
s.stateLock.Lock()
defer s.stateLock.Unlock()
setStateAcquiring := func() {
s.state = contextStateAcquiring
s.contextTaggedLogger.Info("", tag.LifeCycleStarted, tag.ComponentShardContext)
}
s.state = contextStateStopping
s.stopReason = request.reason
}
s.contextTaggedLogger.Info("", tag.LifeCycleStopped, tag.ComponentShardContext)
// Do this again in case we skipped the stopping state, which could happen
// when calling CloseShardByID or the controller is shutting down.
s.lifecycleCancel()
}
case contextStateInitialized:
switch request := request.(type) {
return nil
}
switch request := request.(type) {
case contextRequestAcquire:
return nil // nothing to do, already acquired
setStateStopping(request)
return nil
setStateStopped()
return nil
}
case contextStateStopping:
}
return s.metricsHandler
}
return s.timeSource
}
return s.namespaceRegistry
}
func (s *ContextImpl) GetSearchAttributesProvider() searchattribute.Provider {
}
return s.clusterMetadata
}
return s.archivalMetadata
}
return s.stateMachineRegistry
}
func (s *ContextImpl) ChasmRegistry() *chasm.Registry {
}
return s.endpointRegistry
}
func (s *ContextImpl) BusinessIDReuseRateLimiter(namespaceID namespace.ID, businessID string, archetypeID chasm.ArchetypeID) quotas.RateLimiter {
shard historyi.ShardContext,
logger log.Logger,
return &StateRebuilderImpl{
shard: shard,
namespaceRegistry: shard.GetNamespaceRegistry(),
eventsCache: shard.GetEventsCache(),
clusterMetadata: shard.GetClusterMetadata(),
executionMgr: shard.GetExecutionManager(),
taskRefresher: workflow.NewTaskRefresher(shard),
rebuiltHistorySize: 0,
rebuiltExternalPayloadSize: 0,
rebuiltExternalPayloadCount: 0,
logger: logger,
}
}
func (r *StateRebuilderImpl) Rebuild(
targetBranchToken []byte,
requestID string,
rebuiltMutableState, lastTxnId, err := r.buildMutableStateFromEvent(
ctx,
now,
baseWorkflowIdentifier,
baseBranchToken,
baseLastEventID,
baseLastEventVersion,
targetWorkflowIdentifier,
targetBranchToken,
requestID,
)
if err != nil {
return nil, RebuildStats{}, err
}
// close rebuilt mutable state transaction clearing all generated tasks, etc.
_, _, err = rebuiltMutableState.CloseTransactionAsSnapshot(ctx, historyi.TransactionPolicyPassive)
state_rebuilder.go
if err != nil {
return nil, RebuildStats{}, err
}
// refresh tasks to be generated
// TODO: ideally the executionTimeoutTimerTaskStatus field should be carried over
// from the base run. However, RefreshTasks always resets that field and
// force regenerates the execution timeout timer task.
if err := r.taskRefresher.Refresh(ctx, rebuiltMutableState, false); err != nil {
return nil, RebuildStats{}, err
}
HistorySize: r.rebuiltHistorySize,
ExternalPayloadSize: r.rebuiltExternalPayloadSize,
ExternalPayloadCount: r.rebuiltExternalPayloadCount,
}, nil
}
targetBranchToken []byte,
requestID string,
namespaceEntry, err := r.namespaceRegistry.GetNamespaceByID(namespace.ID(targetWorkflowIdentifier.NamespaceID))
if err != nil {
return nil, 0, err
}
ctx,
common.FirstEventID,
baseLastEventID+1,
baseBranchToken,
namespaceEntry.Name().String(),
))
rebuiltMutableState, stateBuilder := r.initializeBuilders(
namespaceEntry,
targetWorkflowIdentifier,
now,
)
var lastTxnId int64
for iter.HasNext() {
history, err := iter.Next()
switch err.(type) {
case nil:
// noop
case *serviceerror.DataLoss:
}
ctx,
targetWorkflowIdentifier,
stateBuilder,
history.History.Events,
requestID,
); err != nil {
return nil, 0, err
}
}
if err := rebuiltMutableState.SetCurrentBranchToken(targetBranchToken); err != nil {
state_rebuilder.go
return nil, 0, err
}
currentVersionHistory, err := versionhistory.GetCurrentVersionHistory(rebuiltMutableState.GetExecutionInfo().GetVersionHistories())
state_rebuilder.go
if err != nil {
return nil, 0, err
}
lastItem, err := versionhistory.GetLastVersionHistoryItem(currentVersionHistory)
state_rebuilder.go
if err != nil {
return nil, 0, err
}
if !lastItem.Equal(versionhistory.NewVersionHistoryItem(
baseLastEventID,
*baseLastEventVersion,
)) {
return nil, 0, serviceerror.NewInvalidArgumentf(
"StateRebuilder unable to Rebuild mutable state to event ID: %v, version: %v, this event must be at the boundary",
workflowIdentifier definition.WorkflowKey,
now time.Time,
resetMutableState := workflow.NewMutableState(
r.shard,
r.shard.GetEventsCache(),
r.logger,
namespaceEntry,
workflowIdentifier.GetWorkflowID(),
workflowIdentifier.GetRunID(),
now,
)
stateBuilder := workflow.NewMutableStateRebuilder(
r.shard,
r.logger,
resetMutableState,
)
return resetMutableState, stateBuilder
}
func (r *StateRebuilderImpl) applyEvents(
events []*historypb.HistoryEvent,
requestID string,
_, err := stateBuilder.ApplyEvents(
ctx,
namespace.ID(workflowKey.NamespaceID),
requestID,
&commonpb.WorkflowExecution{
WorkflowId: workflowKey.WorkflowID,
RunId: workflowKey.RunID,
},
[][]*historypb.HistoryEvent{events},
nil, // no new run history when rebuilding mutable state
"",
)
if err != nil {
r.logger.Error("StateRebuilder unable to Rebuild mutable state.", tag.Error(err))
return err
}
}
branchToken []byte,
namespaceName string,
return func(paginationToken []byte) ([]HistoryBlobsPaginationItem, []byte, error) {
resp, err := r.executionMgr.ReadHistoryBranchByBatch(ctx, &persistence.ReadHistoryBranchRequest{
BranchToken: branchToken,
MinEventID: firstEventID,
MaxEventID: nextEventID,
PageSize: defaultPageSize,
NextPageToken: paginationToken,
ShardID: r.shard.GetShardID(),
})
if err != nil {
return nil, nil, err
}
paginateItems := make([]HistoryBlobsPaginationItem, 0, len(resp.History))
for i, history := range resp.History {
nextBatch := HistoryBlobsPaginationItem{
History: history,
TransactionID: resp.TransactionIDs[i],
}
paginateItems = append(paginateItems, nextBatch)
// Calculate and accumulate external payload size and count for this batch of history events
if r.shard.GetConfig().ExternalPayloadsEnabled(namespaceName) {
externalPayloadSize, externalPayloadCount, err := workflow.CalculateExternalPayloadSize(
history.Events,
metrics.NoopMetricsHandler, // don't record metrics since those are not new uploads
)
if err != nil {
return nil, nil, err
}
r.rebuiltExternalPayloadCount += externalPayloadCount
}
}
}
}
// NewTest returns a new test resource instance
func NewTest(controller *gomock.Controller, serviceName primitives.ServiceName) *Test {
test_resource.go
logger := log.NewTestLogger()
frontendClient := workflowservicemock.NewMockWorkflowServiceClient(controller)
matchingClient := matchingservicemock.NewMockMatchingServiceClient(controller)
historyClient := historyservicemock.NewMockHistoryServiceClient(controller)
remoteFrontendClient := workflowservicemock.NewMockWorkflowServiceClient(controller)
remoteAdminClient := adminservicemock.NewMockAdminServiceClient(controller)
clusterMetadataManager := persistence.NewMockClusterMetadataManager(controller)
clientBean := client.NewMockBean(controller)
clientBean.EXPECT().GetFrontendClient().Return(frontendClient).AnyTimes()
clientBean.EXPECT().GetMatchingClient(gomock.Any()).Return(matchingClient, nil).AnyTimes()
clientBean.EXPECT().GetHistoryClient().Return(historyClient).AnyTimes()
clientBean.EXPECT().GetRemoteAdminClient(gomock.Any()).Return(remoteAdminClient, nil).AnyTimes()
clientBean.EXPECT().GetRemoteFrontendClient(gomock.Any()).Return(nil, remoteFrontendClient, nil).AnyTimes()
clientFactory := client.NewMockFactory(controller)
metadataMgr := persistence.NewMockMetadataManager(controller)
taskMgr := persistence.NewMockTaskManager(controller)
shardMgr := persistence.NewMockShardManager(controller)
executionMgr := persistence.NewMockExecutionManager(controller)
executionMgr.EXPECT().GetHistoryBranchUtil().Return(persistence.NewHistoryBranchUtil(serialization.NewSerializer())).AnyTimes()
namespaceReplicationQueue := persistence.NewMockNamespaceReplicationQueue(controller)
nexusEndpointMgr := persistence.NewMockNexusEndpointManager(controller)
membershipMonitor := membership.NewMockMonitor(controller)
hostInfoProvider := membership.NewMockHostInfoProvider(controller)
frontendServiceResolver := membership.NewMockServiceResolver(controller)
matchingServiceResolver := membership.NewMockServiceResolver(controller)
historyServiceResolver := membership.NewMockServiceResolver(controller)
workerServiceResolver := membership.NewMockServiceResolver(controller)
membershipMonitor.EXPECT().GetResolver(primitives.FrontendService).Return(frontendServiceResolver, nil).AnyTimes()
membershipMonitor.EXPECT().GetResolver(primitives.InternalFrontendService).Return(nil, membership.ErrUnknownService).AnyTimes()
membershipMonitor.EXPECT().GetResolver(primitives.MatchingService).Return(matchingServiceResolver, nil).AnyTimes()
membershipMonitor.EXPECT().GetResolver(primitives.HistoryService).Return(historyServiceResolver, nil).AnyTimes()
membershipMonitor.EXPECT().GetResolver(primitives.WorkerService).Return(workerServiceResolver, nil).AnyTimes()
membershipMonitor.EXPECT().WaitUntilInitialized(gomock.Any()).Return(nil).AnyTimes()
scope := tally.NewTestScope("test", nil)
metricsHandler := metrics.NewTallyMetricsHandler(metrics.ClientConfig{}, scope).WithTags(
metrics.ServiceNameTag(serviceName),
)
return &Test{
MetricsScope: scope,
ClusterMetadata: cluster.NewMockMetadata(controller),
SearchAttributesProvider: searchattribute.NewMockProvider(controller),
SearchAttributesManager: searchattribute.NewMockManager(controller),
SearchAttributesMapperProvider: searchattribute.NewMockMapperProvider(controller),
// other common resources
NamespaceCache: namespace.NewMockRegistry(controller),
TimeSource: clock.NewRealTimeSource(),
PayloadSerializer: serialization.NewSerializer(),
MetricsHandler: metricsHandler,
ArchivalMetadata: archiver.NewMetadataMock(controller),
ArchiverProvider: provider.NewMockArchiverProvider(controller),
// membership infos
MembershipMonitor: membershipMonitor,
HostInfoProvider: hostInfoProvider,
FrontendServiceResolver: frontendServiceResolver,
MatchingServiceResolver: matchingServiceResolver,
HistoryServiceResolver: historyServiceResolver,
WorkerServiceResolver: workerServiceResolver,
// internal services clients
SDKClientFactory: sdk.NewMockClientFactory(controller),
FrontendClient: frontendClient,
MatchingClient: matchingClient,
HistoryClient: historyClient,
RemoteAdminClient: remoteAdminClient,
RemoteFrontendClient: remoteFrontendClient,
ClientBean: clientBean,
ClientFactory: clientFactory,
ESClient: esclient.NewMockClient(controller),
VisibilityManager: manager.NewMockVisibilityManager(controller),
// persistence clients
MetadataMgr: metadataMgr,
ClusterMetadataMgr: clusterMetadataManager,
TaskMgr: taskMgr,
NamespaceReplicationQueue: namespaceReplicationQueue,
ShardMgr: shardMgr,
ExecutionMgr: executionMgr,
NexusEndpointManager: nexusEndpointMgr,
// logger
Logger: logger,
}
}
// Start for testing
// GetClusterMetadata for testing
return t.ClusterMetadata
}
// GetClusterMetadata for testing
// GetNamespaceRegistry for testing
return t.NamespaceCache
}
// GetTimeSource for testing
// GetPayloadSerializer for testing
return t.PayloadSerializer
}
// GetMetricsHandler for testing
// GetArchivalMetadata for testing
return t.ArchivalMetadata
}
// GetArchiverProvider for testing
// GetHostInfoProvider for testing
return t.HostInfoProvider
}
// GetFrontendServiceResolver for testing
// GetHistoryClient for testing
return t.HistoryClient
}
// GetRemoteAdminClient for testing
// GetClientBean for testing
return t.ClientBean
}
// GetClientFactory for testing
// GetShardManager for testing
return t.ShardMgr
}
// GetExecutionManager for testing
// GetLogger for testing
return t.Logger
}
// GetThrottledLogger for testing
return t.Logger
}
// GetGRPCListener for testing
}
return t.SearchAttributesProvider
}
func (t *Test) GetSearchAttributesManager() searchattribute.Manager {
}
func (t *Test) GetSearchAttributesMapperProvider() searchattribute.MapperProvider {
test_resource.go
return t.SearchAttributesMapperProvider
}
}
if x != nil {
return x.ShardId
}
return 0
}
}
if x != nil {
return x.Owner
}
return ""
}
func (*WorkflowExecutionInfo) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_executions_proto_msgTypes[1]
if x != nil {
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
func (x *WorkflowExecutionInfo) GetVersionHistories() *v14.VersionHistories {
executions.pb.go
if x != nil {
return x.VersionHistories
}
return nil
}
}
func (x *WorkflowExecutionInfo) GetVersioningInfo() *v12.WorkflowExecutionVersioningInfo {
executions.pb.go
if x != nil {
return x.VersioningInfo
}
return nil
}
}
if x != nil {
return x.TimeSkippingInfo
}
return nil
}
func (*TimeSkippingInfo) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_executions_proto_msgTypes[2]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
}
func (x *TimeSkippingInfo) GetAccumulatedSkippedDuration() *durationpb.Duration {
executions.pb.go
if x != nil {
return x.AccumulatedSkippedDuration
}
}
func (*LastNotifiedTargetVersion) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_executions_proto_msgTypes[4]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
func (*ExecutionStats) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_executions_proto_msgTypes[5]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
func (*WorkflowExecutionState) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_executions_proto_msgTypes[6]
if x != nil {
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
func (*RequestIDInfo) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_executions_proto_msgTypes[7]
if x != nil {
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
}
func (*ResetChildInfo) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_executions_proto_msgTypes[28]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
func (*WorkflowPauseInfo) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_executions_proto_msgTypes[29]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
}
func init() { file_temporal_server_api_persistence_v1_executions_proto_init() }
executions.pb.go
func file_temporal_server_api_persistence_v1_executions_proto_init() {
if File_temporal_server_api_persistence_v1_executions_proto != nil {
return
}
file_temporal_server_api_persistence_v1_chasm_proto_init()
file_temporal_server_api_persistence_v1_hsm_proto_init()
file_temporal_server_api_persistence_v1_queues_proto_init()
file_temporal_server_api_persistence_v1_update_proto_init()
file_temporal_server_api_persistence_v1_executions_proto_msgTypes[1].OneofWrappers = []any{
(*WorkflowExecutionInfo_LastWorkflowTaskFailureCause)(nil),
(*WorkflowExecutionInfo_LastWorkflowTaskTimedOutType)(nil),
}
file_temporal_server_api_persistence_v1_executions_proto_msgTypes[8].OneofWrappers = []any{
(*TransferTaskInfo_CloseExecutionTaskDetails_)(nil),
(*TransferTaskInfo_ChasmTaskInfo)(nil),
}
file_temporal_server_api_persistence_v1_executions_proto_msgTypes[10].OneofWrappers = []any{
(*VisibilityTaskInfo_ChasmTaskInfo)(nil),
}
file_temporal_server_api_persistence_v1_executions_proto_msgTypes[11].OneofWrappers = []any{
(*TimerTaskInfo_ChasmTaskInfo)(nil),
}
file_temporal_server_api_persistence_v1_executions_proto_msgTypes[13].OneofWrappers = []any{
(*OutboundTaskInfo_StateMachineInfo)(nil),
(*OutboundTaskInfo_ChasmTaskInfo)(nil),
(*OutboundTaskInfo_WorkerCommandsTask)(nil),
}
file_temporal_server_api_persistence_v1_executions_proto_msgTypes[17].OneofWrappers = []any{
(*ActivityInfo_UseWorkflowBuildIdInfo_)(nil),
(*ActivityInfo_LastIndependentlyAssignedBuildId)(nil),
}
file_temporal_server_api_persistence_v1_executions_proto_msgTypes[23].OneofWrappers = []any{
(*Callback_Nexus_)(nil),
(*Callback_Hsm)(nil),
}
file_temporal_server_api_persistence_v1_executions_proto_msgTypes[40].OneofWrappers = []any{
(*ActivityInfo_PauseInfo_Manual_)(nil),
(*ActivityInfo_PauseInfo_RuleId)(nil),
}
file_temporal_server_api_persistence_v1_executions_proto_msgTypes[46].OneofWrappers = []any{
(*CallbackInfo_Trigger_WorkflowClosed)(nil),
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_persistence_v1_executions_proto_rawDesc), len(file_temporal_server_api_persistence_v1_executions_proto_rawDesc)),
NumEnums: 0,
NumMessages: 47,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_persistence_v1_executions_proto_goTypes,
DependencyIndexes: file_temporal_server_api_persistence_v1_executions_proto_depIdxs,
MessageInfos: file_temporal_server_api_persistence_v1_executions_proto_msgTypes,
}.Build()
File_temporal_server_api_persistence_v1_executions_proto = out.File
file_temporal_server_api_persistence_v1_executions_proto_goTypes = nil
file_temporal_server_api_persistence_v1_executions_proto_depIdxs = nil
}
}
return &sharedScopeCache{
maxSize: maxSize,
scopes: make(map[string]tally.Scope),
handlers: make(map[string]*tallyMetricsHandler),
}
}
func (c *sharedScopeCache) loadOrStoreScope(key string, create func() tally.Scope) tally.Scope {
}
func (c *sharedScopeCache) loadOrStoreHandler(key string, create func() *tallyMetricsHandler) *tallyMetricsHandler {
tally_metrics_handler.go
c.mu.RLock()
if h, ok := c.handlers[key]; ok {
return h
}
h := create()
c.mu.Lock()
defer c.mu.Unlock()
// Double-check: another goroutine may have inserted while we were creating.
if existing, ok := c.handlers[key]; ok {
return existing
}
clear(c.handlers)
}
return h
}
var _ Handler = (*tallyMetricsHandler)(nil)
func NewTallyMetricsHandler(cfg ClientConfig, scope tally.Scope) *tallyMetricsHandler {
tally_metrics_handler.go
perUnitBuckets := make(map[MetricUnit]tally.Buckets)
for unit, boundariesList := range cfg.PerUnitHistogramBoundaries {
perUnitBuckets[MetricUnit(unit)] = tally.ValueBuckets(boundariesList)
}
if maxSize <= 0 {
}
scope: scope,
perUnitBuckets: perUnitBuckets,
excludeTags: configExcludeTags(cfg),
cache: newSharedScopeCache(maxSize),
scopeKey: "",
}
}
// tagsCacheKey builds a compact string key from a tag slice for use as a
// map lookup key.
size := 0
for i := range tags {
size += len(tags[i].Key) + len(tags[i].Value) + 2*binary.MaxVarintLen64
}
var sb strings.Builder
sb.Grow(size)
for _, t := range tags {
appendCacheKeyPart(&sb, t.Key)
appendCacheKeyPart(&sb, t.Value)
}
return sb.String()
}
var lenBuf [binary.MaxVarintLen64]byte
n := binary.PutUvarint(lenBuf[:], uint64(len(value)))
_, _ = sb.Write(lenBuf[:n])
sb.WriteString(value)
}
// WithTags creates a new MetricProvider with provided []Tag
// Tags are merged with registered Tags from the source MetricsHandler.
// Handlers are cached by tag combination so repeated calls avoid allocations.
if len(tags) == 0 {
return tmh
}
normalizedKey := tagsCacheKey(normalizeTagsForCaching(tags, tmh.excludeTags))
tally_metrics_handler.go
key := tmh.scopeKey + normalizedKey
return tmh.cache.loadOrStoreHandler(key, func() *tallyMetricsHandler {
return &tallyMetricsHandler{
scope: tmh.scope.Tagged(tagsToMap(tags, tmh.excludeTags)),
perUnitBuckets: tmh.perUnitBuckets,
excludeTags: tmh.excludeTags,
cache: tmh.cache,
scopeKey: key,
}
})
}
// excludeTags before cache key computation so that different raw values which
// map to the same excluded placeholder share a single cache entry.
func (tmh *tallyMetricsHandler) cachedTaggedScope(tags []Tag) tally.Scope {
tally_metrics_handler.go
if len(tags) == 0 {
}
key := tmh.scopeKey + tagsCacheKey(normalizeTagsForCaching(tags, tmh.excludeTags))
return tmh.cache.loadOrStoreScope(key, func() tally.Scope {
// normalizeTag applies excludeTags substitution to a single tag.
// Returns the (possibly modified) tag and whether it was normalized.
if vals, ok := excl[t.Key]; ok {
if _, ok := vals[t.Value]; !ok {
return Tag{Key: t.Key, Value: tagExcludedValue}, true
}
}
}
// canonical tag values for cache key computation. Returns the original slice
// unchanged if no tags need normalization (zero-alloc fast path).
if len(excl) == 0 {
}
var normalized []Tag
for i, t := range tags {
// Counter obtains a counter for the given name.
func (tmh *tallyMetricsHandler) Counter(counter string) CounterIface {
tally_metrics_handler.go
if v, ok := tmh.counters.Load(counter); ok {
return v.(CounterIface) //nolint:revive // type-safe: only CounterIface is stored
tally_metrics_handler.go
}
tmh.cachedTaggedScope(t).Counter(counter).Inc(i)
})
actual, _ := tmh.counters.LoadOrStore(counter, c)
return actual.(CounterIface) //nolint:revive // type-safe: only CounterIface is stored
}
// Timer obtains a timer for the given name.
if v, ok := tmh.timers.Load(timer); ok {
return v.(TimerIface) //nolint:revive // type-safe: only TimerIface is stored
tally_metrics_handler.go
}
tmh.cachedTaggedScope(t).Timer(timer).Record(d)
})
actual, _ := tmh.timers.LoadOrStore(timer, ti)
return actual.(TimerIface) //nolint:revive // type-safe: only TimerIface is stored
}
// Histogram obtains a histogram for the given name.
func (tmh *tallyMetricsHandler) Histogram(histogram string, unit MetricUnit) HistogramIface {
tally_metrics_handler.go
key := histogramCacheKey{name: histogram, unit: unit}
if v, ok := tmh.histograms.Load(key); ok {
return v.(HistogramIface) //nolint:revive // type-safe: only HistogramIface is stored
}
tmh.cachedTaggedScope(t).Histogram(histogram, tmh.perUnitBuckets[unit]).RecordValue(float64(i))
tally_metrics_handler.go
})
return actual.(HistogramIface) //nolint:revive // type-safe: only HistogramIface is stored
}
}
if len(t1) == 0 {
return nil
}
for i := range t1 {
nt, _ := normalizeTag(t1[i], e)
m[nt.Key] = nt.Value
}
return m
}
shardInfo *persistencespb.ShardInfo,
config *configs.Config,
resourceTest := resourcetest.NewTest(ctrl, primitives.HistoryService)
eventsCache := events.NewMockCache(ctrl)
shard := newTestContext(
resourceTest,
eventsCache,
ContextConfigOverrides{
ShardInfo: shardInfo,
Config: config,
},
)
return &ContextTest{
Resource: resourceTest,
ContextImpl: shard,
MockEventsCache: eventsCache,
}
}
type ContextConfigOverrides struct {
}
func newTestContext(t *resourcetest.Test, eventsCache events.Cache, config ContextConfigOverrides) *ContextImpl {
context_testutil.go
hostInfoProvider := t.GetHostInfoProvider()
lifecycleCtx, lifecycleCancel := context.WithCancel(context.Background())
if config.ShardInfo.QueueStates == nil {
}
if registry == nil {
registry = t.GetNamespaceRegistry()
}
clusterMetadata := config.ClusterMetadata
if clusterMetadata == nil {
clusterMetadata = t.GetClusterMetadata()
}
executionManager := config.ExecutionManager
if executionManager == nil {
executionManager = t.ExecutionMgr
}
taskCategoryRegistry := tasks.NewDefaultTaskCategoryRegistry()
taskCategoryRegistry.AddCategory(tasks.CategoryArchival)
ctx := &ContextImpl{
shardID: config.ShardInfo.GetShardId(),
owner: config.ShardInfo.GetOwner(),
stringRepr: fmt.Sprintf("Shard(%d)", config.ShardInfo.GetShardId()),
executionManager: executionManager,
metricsHandler: t.MetricsHandler,
eventsCache: eventsCache,
config: config.Config,
contextTaggedLogger: t.GetLogger(),
throttledLogger: t.GetThrottledLogger(),
lifecycleCtx: lifecycleCtx,
lifecycleCancel: lifecycleCancel,
queueMetricEmitter: sync.Once{},
state: contextStateAcquired,
engineFuture: future.NewFuture[historyi.Engine](),
shardInfo: config.ShardInfo,
remoteClusterInfos: make(map[string]*remoteClusterInfo),
clusterMetadata: clusterMetadata,
timeSource: t.TimeSource,
namespaceRegistry: registry,
stateMachineRegistry: hsm.NewRegistry(),
chasmRegistry: chasm.NewRegistry(t.GetLogger()),
businessIDRateLimiters: cache.New(
config.Config.BusinessIDReuseLimiterCacheSize(),
&cache.Options{TTL: config.Config.BusinessIDReuseLimiterCacheTTL()},
),
persistenceShardManager: t.GetShardManager(),
clientBean: t.GetClientBean(),
saProvider: t.GetSearchAttributesProvider(),
saMapperProvider: t.GetSearchAttributesMapperProvider(),
historyClient: t.GetHistoryClient(),
payloadSerializer: t.GetPayloadSerializer(),
archivalMetadata: t.GetArchivalMetadata(),
hostInfoProvider: hostInfoProvider,
taskCategoryRegistry: taskCategoryRegistry,
ioSemaphore: locks.NewPrioritySemaphore(1),
}
ctx.taskKeyManager = newTaskKeyManager(
ctx.taskCategoryRegistry,
ctx.timeSource,
config.Config,
ctx.GetLogger(),
func() error {
return ctx.renewRangeLocked(false)
},
)
ctx.handoverTracker = NewDefaultHandoverTrackerFactory()(HandoverTrackerParams{
ClusterMetadata: clusterMetadata,
GetMaxReplicationTaskID: ctx.getMaxReplicationTaskID,
ErrorByStateFn: ctx.errorByState,
NotifyReplicationFn: ctx.notifyReplicationQueueProcessor,
NamespaceRegistry: registry,
Logger: ctx.contextTaggedLogger,
})
return ctx
}
// SetStateMachineRegistry sets the state machine registry on this shard.
s.stateMachineRegistry = reg
}
func (s *ContextTest) SetChasmRegistry(reg *chasm.Registry) {
// should call that, but integration tests need to do it also to clean up any
// background acquireShard goroutines that may exist.
s.FinishStop()
}
func (s *StubContext) GetEngine(_ context.Context) (historyi.Engine, error) {
logger log.Logger,
mutableState historyi.MutableState,
return &MutableStateRebuilderImpl{
shard: shard,
clusterMetadata: shard.GetClusterMetadata(),
namespaceRegistry: shard.GetNamespaceRegistry(),
logger: logger,
mutableState: mutableState,
}
}
func (b *MutableStateRebuilderImpl) ApplyEvents(
newRunHistory []*historypb.HistoryEvent,
newRunID string,
for i := 0; i < len(history)-1; i++ {
_, err := b.applyEvents(ctx, namespaceID, requestID, execution, history[i], nil, "")
if err != nil {
}
}
newMutableState, err := b.applyEvents(ctx, namespaceID, requestID, execution, history[len(history)-1], newRunHistory, newRunID)
mutable_state_rebuilder.go
if err != nil {
return nil, err
}
// close the transaction.
// Previously this comment was here: must generate the activity timer / user timer at the very end
taskGenerator := GetTaskGeneratorProvider().NewTaskGenerator(b.shard, b.mutableState)
mutable_state_rebuilder.go
if err := taskGenerator.GenerateActivityTimerTasks(); err != nil {
return nil, err
}
return nil, err
}
b.mutableState.SetHistoryBuilder(historybuilder.NewImmutable(history...))
mutable_state_rebuilder.go
return newMutableState, nil
}
newRunHistory []*historypb.HistoryEvent,
newRunID string,
if len(history) == 0 {
return nil, serviceerror.NewInternal(ErrMessageHistorySizeZero)
}
lastEvent := history[len(history)-1]
taskGenerator := GetTaskGeneratorProvider().NewTaskGenerator(b.shard, b.mutableState)
// Need to clear the sticky task queue because workflow turned to passive.
b.mutableState.ClearStickyTaskQueue()
executionInfo := b.mutableState.GetExecutionInfo()
executionInfo.LastFirstEventId = firstEvent.GetEventId()
// Preserve the WorkflowTaskStamp during rebuild to ensure workflow task validation works correctly.
// The stamp is used to invalidate stale workflow tasks and must be maintained across rebuilds.
// Note: The stamp is already persisted in the execution info and should not be reset here.
// NOTE: stateRebuilder is also being used in the active side
if err := b.mutableState.UpdateCurrentVersion(lastEvent.GetVersion(), true); err != nil {
return nil, err
}
versionHistory, err := versionhistory.GetCurrentVersionHistory(versionHistories)
if err != nil {
return nil, err
}
if err := versionhistory.AddOrUpdateVersionHistoryItem(versionHistory, versionhistory.NewVersionHistoryItem(
mutable_state_rebuilder.go
lastEvent.GetEventId(),
lastEvent.GetVersion(),
)); err != nil {
return nil, err
}
// [history] is a single persistence batch, so firstEvent.EventId is the batch ID for every
// event applied below. Event definitions that generate load tokens (e.g. NexusOperationScheduled)
// read this via GenerateEventLoadToken to find the original batch ID.
b.mutableState.SetReplayEventBatchID(firstEvent.GetEventId())
for _, event := range history {
switch event.GetEventType() {
attributes := event.GetWorkflowExecutionStartedEventAttributes()
// TODO (alex): ParentWorkflowNamespaceId is back filled. Backward compatibility: old event doesn't have ParentNamespaceId set.
if attributes.GetParentWorkflowNamespaceId() == "" && attributes.GetParentWorkflowNamespace() != "" {
parentNamespaceEntry, err := b.namespaceRegistry.GetNamespace(namespace.Name(attributes.GetParentWorkflowNamespace()))
if err != nil {
}
nil, // shard clock is local to cluster
execution,
requestID,
event,
); err != nil {
return nil, err
}
event,
); err != nil {
return nil, err
}
executionInfo.WorkflowExecutionTimerTaskStatus, err = taskGenerator.GenerateWorkflowStartTasks(
event,
)
if err != nil {
return nil, err
}
if timestamp.DurationValue(attributes.GetFirstWorkflowTaskBackoff()) > 0 {
mutable_state_rebuilder.go
if err := taskGenerator.GenerateDelayedWorkflowTasks(
event,
}
executionInfo.WorkflowExecutionTimeout,
executionInfo.WorkflowRunTimeout,
execution.GetRunId(),
); err != nil {
return nil, err
}
// No mutable state action is needed
if err := b.mutableState.ApplyWorkflowExecutionSignaled(
event,
); err != nil {
return nil, err
}
// The length of newRunHistory can be zero in resend case
return nil, nil
}
if b.mutableState.GetExecutionState().Status == enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING {
}
return SearchAttributeFieldBool{
field: resolveFieldName(enumspb.INDEXED_VALUE_TYPE_BOOL, index),
}
}
// SearchAttributeFieldDateTime is a search attribute field for a datetime value.
}
func newSearchAttributeFieldDateTime(index int) SearchAttributeFieldDateTime {
search_attribute.go
return SearchAttributeFieldDateTime{
field: resolveFieldName(enumspb.INDEXED_VALUE_TYPE_DATETIME, index),
}
}
// SearchAttributeFieldInt is a search attribute field for an integer value.
}
return SearchAttributeFieldInt{
field: resolveFieldName(enumspb.INDEXED_VALUE_TYPE_INT, index),
}
}
// SearchAttributeFieldDouble is a search attribute field for a double value.
}
func newSearchAttributeFieldDouble(index int) SearchAttributeFieldDouble {
search_attribute.go
return SearchAttributeFieldDouble{
field: resolveFieldName(enumspb.INDEXED_VALUE_TYPE_DOUBLE, index),
}
}
// SearchAttributeFieldKeyword is a search attribute field for a keyword value.
}
func newSearchAttributeFieldKeyword(index int) SearchAttributeFieldKeyword {
search_attribute.go
return SearchAttributeFieldKeyword{
field: resolveFieldName(enumspb.INDEXED_VALUE_TYPE_KEYWORD, index),
}
}
func newSearchAttributeFieldLowCardinalityKeyword(index int) SearchAttributeFieldKeyword {
search_attribute.go
return SearchAttributeFieldKeyword{
field: fmt.Sprintf("%s%s%02d", sadefs.ReservedPrefix, "LowCardinalityKeyword", index),
}
}
// SearchAttributeFieldKeywordList is a search attribute field for a keyword list value.
}
func newSearchAttributeFieldKeywordList(index int) SearchAttributeFieldKeywordList {
search_attribute.go
return SearchAttributeFieldKeywordList{
field: resolveFieldName(enumspb.INDEXED_VALUE_TYPE_KEYWORD_LIST, index),
}
}
// SearchAttributeFieldText is a search attribute field for a text value.
}
func resolveFieldName(valueType enumspb.IndexedValueType, index int) string {
search_attribute.go
// Columns are named like TemporalBool01, TemporalDatetime01, TemporalDouble01, TemporalInt01.
return fmt.Sprintf("%s%s%02d", sadefs.ReservedPrefix, valueType.String(), index)
}
func (s searchAttributeDefinition) definition() searchAttributeDefinition {
}
return SearchAttributeBool{
searchAttributeDefinition: searchAttributeDefinition{
alias: field,
field: field,
valueType: enumspb.INDEXED_VALUE_TYPE_BOOL,
},
}
}
// Value sets the boolean value of the search attribute.
}
func newSearchAttributeDateTimeByField(field string) SearchAttributeDateTime {
search_attribute.go
return SearchAttributeDateTime{
searchAttributeDefinition: searchAttributeDefinition{
alias: field,
field: field,
valueType: enumspb.INDEXED_VALUE_TYPE_DATETIME,
},
}
}
// Value sets the date time value of the search attribute.
// NewSearchAttributeKeyword creates a new keyword search attribute given a predefined chasm field
func NewSearchAttributeKeyword(alias string, keywordField SearchAttributeFieldKeyword) SearchAttributeKeyword {
search_attribute.go
return SearchAttributeKeyword{
searchAttributeDefinition: searchAttributeDefinition{
alias: alias,
field: keywordField.field,
valueType: enumspb.INDEXED_VALUE_TYPE_KEYWORD,
},
}
}
func newSearchAttributeKeywordByField(field string) SearchAttributeKeyword {
search_attribute.go
return SearchAttributeKeyword{
searchAttributeDefinition: searchAttributeDefinition{
alias: field,
field: field,
valueType: enumspb.INDEXED_VALUE_TYPE_KEYWORD,
},
}
}
// Value sets the string value of the search attribute.
}
func newSearchAttributeKeywordListByField(field string) SearchAttributeKeywordList {
search_attribute.go
return SearchAttributeKeywordList{
searchAttributeDefinition: searchAttributeDefinition{
alias: field,
field: field,
valueType: enumspb.INDEXED_VALUE_TYPE_KEYWORD_LIST,
},
}
}
// Value sets the string list value of the search attribute.
func (*Predicate) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[0]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
func (*UniversalPredicateAttributes) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[1]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
func (*EmptyPredicateAttributes) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[2]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
func (*AndPredicateAttributes) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[3]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
func (*OrPredicateAttributes) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[4]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
func (*NotPredicateAttributes) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[5]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
func (*NamespaceIdPredicateAttributes) ProtoMessage() {}
func (x *NamespaceIdPredicateAttributes) ProtoReflect() protoreflect.Message {
predicates.pb.go
mi := &file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[6]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
func (*TaskTypePredicateAttributes) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[7]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
func (*DestinationPredicateAttributes) ProtoMessage() {}
func (x *DestinationPredicateAttributes) ProtoReflect() protoreflect.Message {
predicates.pb.go
mi := &file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[8]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
func (*OutboundTaskGroupPredicateAttributes) ProtoMessage() {}
func (x *OutboundTaskGroupPredicateAttributes) ProtoReflect() protoreflect.Message {
predicates.pb.go
mi := &file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[9]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
func (*OutboundTaskPredicateAttributes) ProtoMessage() {}
func (x *OutboundTaskPredicateAttributes) ProtoReflect() protoreflect.Message {
predicates.pb.go
mi := &file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[10]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
}
func init() { file_temporal_server_api_persistence_v1_predicates_proto_init() }
predicates.pb.go
func file_temporal_server_api_persistence_v1_predicates_proto_init() {
if File_temporal_server_api_persistence_v1_predicates_proto != nil {
return
}
file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[0].OneofWrappers = []any{
(*Predicate_UniversalPredicateAttributes)(nil),
(*Predicate_EmptyPredicateAttributes)(nil),
(*Predicate_AndPredicateAttributes)(nil),
(*Predicate_OrPredicateAttributes)(nil),
(*Predicate_NotPredicateAttributes)(nil),
(*Predicate_NamespaceIdPredicateAttributes)(nil),
(*Predicate_TaskTypePredicateAttributes)(nil),
(*Predicate_DestinationPredicateAttributes)(nil),
(*Predicate_OutboundTaskGroupPredicateAttributes)(nil),
(*Predicate_OutboundTaskPredicateAttributes)(nil),
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_persistence_v1_predicates_proto_rawDesc), len(file_temporal_server_api_persistence_v1_predicates_proto_rawDesc)),
NumEnums: 0,
NumMessages: 12,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_persistence_v1_predicates_proto_goTypes,
DependencyIndexes: file_temporal_server_api_persistence_v1_predicates_proto_depIdxs,
MessageInfos: file_temporal_server_api_persistence_v1_predicates_proto_msgTypes,
}.Build()
File_temporal_server_api_persistence_v1_predicates_proto = out.File
file_temporal_server_api_persistence_v1_predicates_proto_goTypes = nil
file_temporal_server_api_persistence_v1_predicates_proto_depIdxs = nil
}
// NewTestLogger returns a logger for tests
// Deprecated: Use testlogger.TestLogger instead.
format := os.Getenv(TestLogFormatEnvVar)
if format == "" {
format = "console"
}
Level: os.Getenv(TestLogLevelEnvVar),
Format: format,
Development: true,
})
// Don't include stack traces for warnings during tests. Only include them for logs with level error and above.
logger = logger.WithOptions(zap.AddStacktrace(zap.ErrorLevel))
return NewZapLogger(logger)
}
// NewZapLogger returns a new zap based logger from zap.Logger
return &zapLogger{
zl: zl,
skip: skipForZapLogger,
baseZl: zl,
}
}
// BuildZapLogger builds and returns a new zap.Logger for this logging configuration
return buildZapLogger(cfg, true)
}
_, path, line, ok := runtime.Caller(skip)
if !ok {
return ""
}
}
fields := make([]zap.Field, len(tags)+1)
l.fillFields(tags, fields)
fields[len(fields)-1] = zap.String(tag.LoggingCallAtKey, caller(l.skip))
return fields
}
// fillFields fill fields parameter with fields read from tags. Optimized for performance.
for i, t := range tags {
fields[i] = zt.Field()
} else {
fields[i] = zap.Any(t.Key(), t.Value())
}
}
if msg == "" {
}
}
}
if l.zl.Core().Enabled(zap.InfoLevel) {
msg = setDefaultMsg(msg)
fields := l.buildFieldsWithCallAt(tags)
l.zl.Info(msg, fields...)
}
}
}
encodeConfig := DefaultZapEncoderConfig
if disableCaller {
encodeConfig.CallerKey = zapcore.OmitKey
encodeConfig.EncodeCaller = nil
}
if len(cfg.OutputFile) > 0 {
outputPath = cfg.OutputFile
}
outputPath = "stdout"
}
if cfg.Format == "console" {
}
Level: zap.NewAtomicLevelAt(ParseZapLevel(cfg.Level)),
Development: cfg.Development,
Sampling: nil,
Encoding: encoding,
EncoderConfig: encodeConfig,
OutputPaths: []string{outputPath},
ErrorOutputPaths: []string{outputPath},
DisableCaller: disableCaller,
}
logger, _ := config.Build()
return logger
}
}
switch strings.ToLower(level) {
case "debug":
return zap.DebugLevel
case "fatal":
return zap.FatalLevel
return zap.InfoLevel
}
}
archivalMetadata archiver.ArchivalMetadata,
logger log.Logger,
return &TaskGeneratorImpl{
namespaceRegistry: namespaceRegistry,
mutableState: mutableState,
config: config,
archivalMetadata: archivalMetadata,
logger: logger,
}
}
func (r *TaskGeneratorImpl) GenerateWorkflowStartTasks(
startEvent *historypb.HistoryEvent,
executionInfo := r.mutableState.GetExecutionInfo()
executionTimeoutTimerTaskStatus := executionInfo.WorkflowExecutionTimerTaskStatus
if !r.mutableState.IsWorkflowExecutionRunning() {
return executionTimeoutTimerTaskStatus, nil
}
workflowExecutionTimeoutTimerEnabled := r.config.EnableWorkflowExecutionTimeoutTimer()
task_generator.go
if !workflowExecutionTimeoutTimerEnabled {
// when the feature is disabled, reset this field so that it won't be carried over to the next run
// and new runs can always have the run timeout timer always generated.
// into the situation where execution timeout is set but no timeout timer task is generated.
isFirstRun := executionInfo.FirstExecutionRunId == r.mutableState.GetExecutionState().RunId
task_generator.go
workflowExecutionExpirationTime := timestamp.TimeValue(
executionInfo.WorkflowExecutionExpirationTime,
)
if workflowExecutionTimeoutTimerEnabled &&
!isFirstRun &&
!workflowExecutionExpirationTime.IsZero() &&
executionInfo.WorkflowExecutionTimerTaskStatus == TimerTaskStatusNone {
r.mutableState.AddTasks(&tasks.WorkflowExecutionTimeoutTask{
// TaskID is set by shard
}
executionInfo.WorkflowRunExpirationTime,
)
if workflowRunExpirationTime.IsZero() {
return executionTimeoutTimerTaskStatus, nil
}
workflowRunExpirationTime.Before(workflowExecutionExpirationTime) {
// TaskID is set by shard
WorkflowKey: r.mutableState.GetWorkflowKey(),
VisibilityTimestamp: workflowRunExpirationTime,
Version: startEvent.GetVersion(),
})
}
}
func (r *TaskGeneratorImpl) GenerateDirtySubStateMachineTasks(
stateMachineRegistry *hsm.Registry,
tree := r.mutableState.HSM()
opLog, err := tree.OpLog()
if err != nil {
return err
}
switch transitionOp := op.(type) {
case hsm.DeleteOperation:
}
return nil
}
func (r *TaskGeneratorImpl) GenerateRecordWorkflowStartedTasks(
startEvent *historypb.HistoryEvent,
startVersion := startEvent.GetVersion()
r.mutableState.AddTasks(&tasks.StartExecutionVisibilityTask{
// TaskID, VisibilityTimestamp is set by shard
WorkflowKey: r.mutableState.GetWorkflowKey(),
Version: startVersion,
})
return nil
}
func (r *TaskGeneratorImpl) GenerateScheduleWorkflowTaskTasks(
}
_, err := r.getTimerSequence().CreateNextActivityTimer()
return err
}
_, err := r.getTimerSequence().CreateNextUserTimer()
return err
}
func (r *TaskGeneratorImpl) GenerateHistoryReplicationTasks(
}
return NewTimerSequence(r.mutableState)
}
func (r *TaskGeneratorImpl) getTargetNamespaceID(
func (*VersionHistoryItem) ProtoMessage() {}
mi := &file_temporal_server_api_history_v1_message_proto_msgTypes[1]
if x != nil {
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
}
}
if x != nil {
}
return 0
}
if x != nil {
}
return 0
}
func (*VersionHistory) ProtoMessage() {}
mi := &file_temporal_server_api_history_v1_message_proto_msgTypes[2]
if x != nil {
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
}
}
if x != nil {
return x.BranchToken
}
return nil
}
func (*VersionHistories) ProtoMessage() {}
mi := &file_temporal_server_api_history_v1_message_proto_msgTypes[3]
if x != nil {
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
}
}
if x != nil {
return x.CurrentVersionHistoryIndex
}
return 0
}
}
func file_temporal_server_api_history_v1_message_proto_init() {
if File_temporal_server_api_history_v1_message_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_history_v1_message_proto_rawDesc), len(file_temporal_server_api_history_v1_message_proto_rawDesc)),
NumEnums: 0,
NumMessages: 8,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_history_v1_message_proto_goTypes,
DependencyIndexes: file_temporal_server_api_history_v1_message_proto_depIdxs,
MessageInfos: file_temporal_server_api_history_v1_message_proto_msgTypes,
}.Build()
File_temporal_server_api_history_v1_message_proto = out.File
file_temporal_server_api_history_v1_message_proto_goTypes = nil
file_temporal_server_api_history_v1_message_proto_depIdxs = nil
}
}
return len(b.memEventsBatches) > 0 ||
len(b.memLatestBatch) > 0 ||
len(b.memBufferBatch) > 0 ||
len(b.scheduledIDToStartedID) > 0
}
func (b *EventStore) AllocateEventID() int64 {
}
return b.nextEventID
}
func (b *EventStore) LastEventVersion() (int64, bool) {
}
func (b *EventStore) FlushBufferToCurrentBatch() (map[int64]int64, map[string]int64) {
event_store.go
if len(b.dbBufferBatch) == 0 && len(b.memBufferBatch) == 0 {
}
b.assertMutable()
}
b.assertNotSealed()
if len(b.memLatestBatch) == 0 {
}
b.memEventsBatches = append(b.memEventsBatches, b.memLatestBatch)
func (b *EventStore) Finish(
flushBufferEvent bool,
defer func() {
b.state = HistoryBuilderStateSealed
}()
}
dbEventsBatches := b.memEventsBatches
dbClearBuffer := b.dbClearBuffer
dbBufferBatch := b.memBufferBatch
memBufferBatch := b.dbBufferBatch
memBufferBatch = append(memBufferBatch, dbBufferBatch...)
scheduledIDToStartedID := b.scheduledIDToStartedID
requestIDToEventID := b.requestIDToEventID
b.memEventsBatches = nil
b.memBufferBatch = nil
b.memLatestBatch = nil
b.memLatestBatchSize = 0
b.dbClearBuffer = false
b.dbBufferBatch = nil
b.scheduledIDToStartedID = nil
if err := b.assignTaskIDs(dbEventsBatches); err != nil {
return nil, err
}
DBEventsBatches: dbEventsBatches,
DBClearBuffer: dbClearBuffer,
DBBufferBatch: dbBufferBatch,
MemBufferBatch: memBufferBatch,
ScheduledIDToStartedID: scheduledIDToStartedID,
RequestIDToEventID: requestIDToEventID,
}, nil
}
func (b *EventStore) assignTaskIDs(
dbEventsBatches [][]*historypb.HistoryEvent,
b.assertNotSealed()
if b.state == HistoryBuilderStateImmutable {
}
taskIDCount := 0
}
if b.state == HistoryBuilderStateSealed {
panic("history builder is in sealed state")
}
metricsHandler metrics.Handler,
maxEventBatchSizeInBytes dynamicconfig.IntPropertyFn,
return &HistoryBuilder{
EventStore: EventStore{
state: HistoryBuilderStateMutable,
timeSource: timeSource,
taskIDGenerator: taskIDGenerator,
version: version,
nextEventID: nextEventID,
workflowFinished: false,
dbBufferBatch: dbBufferBatch,
dbClearBuffer: false,
memEventsBatches: nil,
memLatestBatch: nil,
memBufferBatch: nil,
scheduledIDToStartedID: make(map[int64]int64),
requestIDToEventID: make(map[string]int64),
maxEventBatchSizeInBytes: maxEventBatchSizeInBytes,
metricsHandler: metricsHandler,
},
EventFactory: EventFactory{timeSource: timeSource, version: version},
}
}
func (b *HistoryBuilder) SetTimeSource(timeSource clock.TimeSource) {
}
func NewImmutable(histories ...[]*historypb.HistoryEvent) *HistoryBuilder {
history_builder.go
lastHistory := histories[len(histories)-1]
lastEvent := lastHistory[len(lastHistory)-1]
return &HistoryBuilder{
EventStore: EventStore{
state: HistoryBuilderStateImmutable,
timeSource: nil,
taskIDGenerator: nil,
version: lastEvent.GetVersion(),
nextEventID: lastEvent.GetEventId() + 1,
workflowFinished: false,
dbBufferBatch: nil,
dbClearBuffer: false,
memEventsBatches: histories,
memLatestBatch: nil,
memBufferBatch: nil,
scheduledIDToStartedID: nil,
requestIDToEventID: nil,
metricsHandler: nil,
},
EventFactory: EventFactory{},
}
}
func NewImmutableForUpdateNextEventID(lastVersionHistoryItem *historyspb.VersionHistoryItem) *HistoryBuilder {
}
return b.EventStore.IsDirty()
}
// AddWorkflowExecutionStartedEvent
func (*StateMachineMap) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_hsm_proto_msgTypes[1]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
func (*StateMachineRef) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_hsm_proto_msgTypes[3]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
func (*StateMachineTaskInfo) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_hsm_proto_msgTypes[4]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
func (*StateMachineTimerGroup) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_hsm_proto_msgTypes[5]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
func (*VersionedTransition) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_hsm_proto_msgTypes[6]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
func (*StateMachineTombstoneBatch) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_hsm_proto_msgTypes[7]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
}
func file_temporal_server_api_persistence_v1_hsm_proto_init() {
if File_temporal_server_api_persistence_v1_hsm_proto != nil {
return
}
file_temporal_server_api_persistence_v1_hsm_proto_msgTypes[8].OneofWrappers = []any{
(*StateMachineTombstone_ActivityScheduledEventId)(nil),
(*StateMachineTombstone_TimerId)(nil),
(*StateMachineTombstone_ChildExecutionInitiatedEventId)(nil),
(*StateMachineTombstone_RequestCancelInitiatedEventId)(nil),
(*StateMachineTombstone_SignalExternalInitiatedEventId)(nil),
(*StateMachineTombstone_UpdateId)(nil),
(*StateMachineTombstone_StateMachinePath)(nil),
(*StateMachineTombstone_ChasmNodePath)(nil),
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_persistence_v1_hsm_proto_rawDesc), len(file_temporal_server_api_persistence_v1_hsm_proto_rawDesc)),
NumEnums: 0,
NumMessages: 12,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_persistence_v1_hsm_proto_goTypes,
DependencyIndexes: file_temporal_server_api_persistence_v1_hsm_proto_depIdxs,
MessageInfos: file_temporal_server_api_persistence_v1_hsm_proto_msgTypes,
}.Build()
File_temporal_server_api_persistence_v1_hsm_proto = out.File
file_temporal_server_api_persistence_v1_hsm_proto_goTypes = nil
file_temporal_server_api_persistence_v1_hsm_proto_depIdxs = nil
}
// NewMockShardManager creates a new mock instance.
mock := &MockShardManager{ctrl: ctrl}
mock.recorder = &MockShardManagerMockRecorder{mock}
return mock
}
// EXPECT returns an object that allows the caller to indicate expected use.
// NewMockExecutionManager creates a new mock instance.
func NewMockExecutionManager(ctrl *gomock.Controller) *MockExecutionManager {
data_interfaces_mock.go
mock := &MockExecutionManager{ctrl: ctrl}
mock.recorder = &MockExecutionManagerMockRecorder{mock}
return mock
}
// EXPECT returns an object that allows the caller to indicate expected use.
func (m *MockExecutionManager) EXPECT() *MockExecutionManagerMockRecorder {
data_interfaces_mock.go
return m.recorder
}
// AddHistoryTasks mocks base method.
// GetHistoryBranchUtil mocks base method.
func (m *MockExecutionManager) GetHistoryBranchUtil() HistoryBranchUtil {
data_interfaces_mock.go
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "GetHistoryBranchUtil")
ret0, _ := ret[0].(HistoryBranchUtil)
return ret0
}
// GetHistoryBranchUtil indicates an expected call of GetHistoryBranchUtil.
func (mr *MockExecutionManagerMockRecorder) GetHistoryBranchUtil() *gomock.Call {
data_interfaces_mock.go
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetHistoryBranchUtil", reflect.TypeOf((*MockExecutionManager)(nil).GetHistoryBranchUtil))
}
// GetHistoryTasks mocks base method.
// ReadHistoryBranchByBatch mocks base method.
func (m *MockExecutionManager) ReadHistoryBranchByBatch(ctx context.Context, request *ReadHistoryBranchRequest) (*ReadHistoryBranchByBatchResponse, error) {
data_interfaces_mock.go
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "ReadHistoryBranchByBatch", ctx, request)
ret0, _ := ret[0].(*ReadHistoryBranchByBatchResponse)
ret1, _ := ret[1].(error)
return ret0, ret1
}
// ReadHistoryBranchByBatch indicates an expected call of ReadHistoryBranchByBatch.
func (mr *MockExecutionManagerMockRecorder) ReadHistoryBranchByBatch(ctx, request any) *gomock.Call {
data_interfaces_mock.go
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ReadHistoryBranchByBatch", reflect.TypeOf((*MockExecutionManager)(nil).ReadHistoryBranchByBatch), ctx, request)
}
// ReadHistoryBranchReverse mocks base method.
// NewMockTaskManager creates a new mock instance.
mock := &MockTaskManager{ctrl: ctrl}
mock.recorder = &MockTaskManagerMockRecorder{mock}
return mock
}
// EXPECT returns an object that allows the caller to indicate expected use.
// NewMockMetadataManager creates a new mock instance.
func NewMockMetadataManager(ctrl *gomock.Controller) *MockMetadataManager {
data_interfaces_mock.go
mock := &MockMetadataManager{ctrl: ctrl}
mock.recorder = &MockMetadataManagerMockRecorder{mock}
return mock
}
// EXPECT returns an object that allows the caller to indicate expected use.
// NewMockClusterMetadataManager creates a new mock instance.
func NewMockClusterMetadataManager(ctrl *gomock.Controller) *MockClusterMetadataManager {
data_interfaces_mock.go
mock := &MockClusterMetadataManager{ctrl: ctrl}
mock.recorder = &MockClusterMetadataManagerMockRecorder{mock}
return mock
}
// EXPECT returns an object that allows the caller to indicate expected use.
// NewMockNexusEndpointManager creates a new mock instance.
func NewMockNexusEndpointManager(ctrl *gomock.Controller) *MockNexusEndpointManager {
data_interfaces_mock.go
mock := &MockNexusEndpointManager{ctrl: ctrl}
mock.recorder = &MockNexusEndpointManagerMockRecorder{mock}
return mock
}
// EXPECT returns an object that allows the caller to indicate expected use.
children map[string]*persistencespb.StateMachineMap,
backend NodeBackend,
def, ok := registry.Machine(t)
if !ok {
return nil, fmt.Errorf("%w: state machine for type: %v", ErrNotRegistered, t)
}
if err != nil {
return nil, err
}
definition: def,
registry: registry,
persistence: &persistencespb.StateMachineNode{
Children: children,
Data: serialized,
InitialVersionedTransition: &persistencespb.VersionedTransition{},
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{},
TransitionCount: 0,
},
cache: &cachedMachine{
dataLoaded: true,
data: data,
children: make(map[Key]*Node),
},
backend: backend,
opLog: make(OperationLog, 0),
}, nil
}
// deleted. For details on compaction rules, see OperationLog.compact().
// This method must be called on the root node only.
if n.Parent != nil {
return nil, fmt.Errorf("can only be called from root node")
}
return compacted, nil
}
// This should be called at the end of every transaction where the transitions are performed to avoid emitting duplicate
// transition outputs.
n.root().opLog = nil
n.cache.dirty = false
for _, child := range n.cache.children {
child.ClearTransactionState()
}
// Walk applies the given function to all nodes rooted at the current node.
// Returns after successfully applying the function to all nodes or first error.
if n == nil {
return nil
}
return err
}
childNodes := NewCollection[any](n, childType).List()
for _, child := range childNodes {
// InternalRepr returns the internal persistence representation of this node.
// Meant to be used by the framework, **not** by components.
return n.persistence
}
// CompareState compare current node state with the incoming node state.
// NewCollection creates a new collection. For subscriptions to work, you must call Start/Stop.
// Get will work without Start/Stop.
// Do this at the first convenient place we have a logger:
logSharedStructureWarnings(logger)
return &Collection{
client: client,
logger: logger,
errCount: -1,
subscriptions: make(map[Key]map[int]any),
convertCache: new(sync.Map),
indexCache: new(sync.Map),
}
}
func (c *Collection) Start() {
cvs []ConstrainedValue,
precedence []Constraints,
if len(cvs) == 0 {
return findMatchWithCache(cache, cvs, precedence)
}
convert func(value any) (T, error),
precedence []Constraints,
cvs := c.client.GetValue(key)
v, _ := matchAndConvertCvs(c, key, def, convert, precedence, cvs)
return v
}
func matchAndConvertCvs[T any](
precedence []Constraints,
cvs []ConstrainedValue,
cvp, err := findMatch(c.indexCache, cvs, precedence)
if err != nil {
return def, usingDefaultValue
}
typedVal, err := convertWithCache(c, key, convert, cvp)
// treat the fields independently), or the zero value of its type (if you want to treat the fields
// as a group and default unset fields to zero).
return func(v any) (T, error) {
// if we already have the right type, no conversion is necessary
if typedV, ok := v.(T); ok {
return typedV, nil
}
// Deep-copy the default and decode over it. This allows using e.g. a struct with some
// default fields filled in and a config that only set some fields.
dec, err := mapstructure.NewDecoder(&mapstructure.DecoderConfig{
Result: &out,
DecodeHook: mapstructure.ComposeDecodeHookFunc(
mapstructureHookDuration,
mapstructureHookTimestamp,
mapstructureHookProtoEnum,
mapstructureHookGeneric,
),
})
if err != nil {
return out, err
}
return out, err
}
}
}
func init() { file_temporal_server_api_historyservice_v1_request_response_proto_init() }
request_response.pb.go
func file_temporal_server_api_historyservice_v1_request_response_proto_init() {
if File_temporal_server_api_historyservice_v1_request_response_proto != nil {
return
}
file_temporal_server_api_historyservice_v1_request_response_proto_msgTypes[107].OneofWrappers = []any{
(*StreamWorkflowReplicationMessagesRequest_SyncReplicationState)(nil),
}
file_temporal_server_api_historyservice_v1_request_response_proto_msgTypes[108].OneofWrappers = []any{
(*StreamWorkflowReplicationMessagesResponse_Messages)(nil),
}
file_temporal_server_api_historyservice_v1_request_response_proto_msgTypes[134].OneofWrappers = []any{
(*CompleteNexusOperationChasmRequest_Success)(nil),
(*CompleteNexusOperationChasmRequest_Failure)(nil),
}
file_temporal_server_api_historyservice_v1_request_response_proto_msgTypes[136].OneofWrappers = []any{
(*CompleteNexusOperationRequest_Success)(nil),
(*CompleteNexusOperationRequest_Failure)(nil),
}
file_temporal_server_api_historyservice_v1_request_response_proto_msgTypes[162].OneofWrappers = []any{
(*ExecuteMultiOperationRequest_Operation_StartWorkflow)(nil),
(*ExecuteMultiOperationRequest_Operation_UpdateWorkflow)(nil),
}
file_temporal_server_api_historyservice_v1_request_response_proto_msgTypes[163].OneofWrappers = []any{
(*ExecuteMultiOperationResponse_Response_StartWorkflow)(nil),
(*ExecuteMultiOperationResponse_Response_UpdateWorkflow)(nil),
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_historyservice_v1_request_response_proto_rawDesc), len(file_temporal_server_api_historyservice_v1_request_response_proto_rawDesc)),
NumEnums: 0,
NumMessages: 171,
NumExtensions: 1,
NumServices: 0,
},
GoTypes: file_temporal_server_api_historyservice_v1_request_response_proto_goTypes,
DependencyIndexes: file_temporal_server_api_historyservice_v1_request_response_proto_depIdxs,
MessageInfos: file_temporal_server_api_historyservice_v1_request_response_proto_msgTypes,
ExtensionInfos: file_temporal_server_api_historyservice_v1_request_response_proto_extTypes,
}.Build()
File_temporal_server_api_historyservice_v1_request_response_proto = out.File
file_temporal_server_api_historyservice_v1_request_response_proto_goTypes = nil
file_temporal_server_api_historyservice_v1_request_response_proto_depIdxs = nil
}
)
// This line cannot be combined with others!
// This ensures the stack trace has information of the caller.
def := newMetricDefinition(name, opts...)
globalRegistry.register(def)
return timerDefinition{def}
}
// This line cannot be combined with others!
// This ensures the stack trace has information of the caller.
def := newMetricDefinition(name, append(opts, WithUnit(Bytes))...)
globalRegistry.register(def)
return histogramDefinition{def}
}
// This line cannot be combined with others!
// This ensures the stack trace has information of the caller.
def := newMetricDefinition(name, append(opts, WithUnit(Dimensionless))...)
globalRegistry.register(def)
return histogramDefinition{def}
}
// This line cannot be combined with others!
// This ensures the stack trace has information of the caller.
def := newMetricDefinition(name, opts...)
globalRegistry.register(def)
return counterDefinition{def}
}
// This line cannot be combined with others!
// This ensures the stack trace has information of the caller.
def := newMetricDefinition(name, opts...)
globalRegistry.register(def)
return gaugeDefinition{def}
}
func (d histogramDefinition) With(handler Handler) HistogramIface {
}
return handler.Counter(d.name)
}
return handler.Gauge(d.name)
}
return handler.Timer(d.name)
}
}
func init() { file_temporal_server_api_matchingservice_v1_request_response_proto_init() }
request_response.pb.go
func file_temporal_server_api_matchingservice_v1_request_response_proto_init() {
if File_temporal_server_api_matchingservice_v1_request_response_proto != nil {
return
}
file_temporal_server_api_matchingservice_v1_request_response_proto_msgTypes[27].OneofWrappers = []any{
(*UpdateWorkerBuildIdCompatibilityRequest_ApplyPublicRequest_)(nil),
(*UpdateWorkerBuildIdCompatibilityRequest_RemoveBuildIds_)(nil),
(*UpdateWorkerBuildIdCompatibilityRequest_PersistUnknownBuildId)(nil),
}
file_temporal_server_api_matchingservice_v1_request_response_proto_msgTypes[29].OneofWrappers = []any{
(*GetWorkerVersioningRulesRequest_Request)(nil),
}
file_temporal_server_api_matchingservice_v1_request_response_proto_msgTypes[31].OneofWrappers = []any{
(*UpdateWorkerVersioningRulesRequest_Request)(nil),
}
file_temporal_server_api_matchingservice_v1_request_response_proto_msgTypes[37].OneofWrappers = []any{
(*SyncDeploymentUserDataRequest_UpdateVersionData)(nil),
(*SyncDeploymentUserDataRequest_ForgetVersion)(nil),
}
file_temporal_server_api_matchingservice_v1_request_response_proto_msgTypes[56].OneofWrappers = []any{
(*DispatchNexusTaskResponse_HandlerError)(nil),
(*DispatchNexusTaskResponse_Response)(nil),
(*DispatchNexusTaskResponse_RequestTimeout)(nil),
(*DispatchNexusTaskResponse_Failure)(nil),
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_matchingservice_v1_request_response_proto_rawDesc), len(file_temporal_server_api_matchingservice_v1_request_response_proto_rawDesc)),
NumEnums: 0,
NumMessages: 97,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_matchingservice_v1_request_response_proto_goTypes,
DependencyIndexes: file_temporal_server_api_matchingservice_v1_request_response_proto_depIdxs,
MessageInfos: file_temporal_server_api_matchingservice_v1_request_response_proto_msgTypes,
}.Build()
File_temporal_server_api_matchingservice_v1_request_response_proto = out.File
file_temporal_server_api_matchingservice_v1_request_response_proto_goTypes = nil
file_temporal_server_api_matchingservice_v1_request_response_proto_depIdxs = nil
}
func NewPagingIterator[V any](
paginationFn PaginationFn[V],
iter := &PagingIteratorImpl[V]{
paginationFn: paginationFn,
pageToken: nil,
pageErr: nil,
pageItems: nil,
nextPageItemIndex: 0,
}
iter.getNextPage() // this will initialize the paging iterator
return iter
}
// NewPagingIteratorWithToken create a new paging iterator with initial token
// HasNext return whether has next item or err
// pagination encounters error
if iter.pageErr != nil {
return true
}
// still have local cached item to return
}
return iter.HasNext()
}
}
// Next return next item or err
if !iter.HasNext() {
panic("HistoryEventIterator Next() called without checking HasNext()")
}
err := iter.pageErr
iter.pageErr = nil
// we have cached events
index := iter.nextPageItemIndex
iter.nextPageItemIndex++
return iter.pageItems[index], nil
}
panic("HistoryEventIterator Next() should return either a history event or a err")
}
items, token, err := iter.paginationFn(iter.pageToken)
if err == nil {
iter.pageToken = token
iter.pageErr = nil
iter.pageItems = nil
iter.pageToken = nil
iter.pageErr = err
}
}
func (*ChasmTaskInfo) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_chasm_proto_msgTypes[8]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
}
func file_temporal_server_api_persistence_v1_chasm_proto_init() {
if File_temporal_server_api_persistence_v1_chasm_proto != nil {
return
}
file_temporal_server_api_persistence_v1_hsm_proto_init()
file_temporal_server_api_persistence_v1_chasm_proto_msgTypes[1].OneofWrappers = []any{
(*ChasmNodeMetadata_ComponentAttributes)(nil),
(*ChasmNodeMetadata_DataAttributes)(nil),
(*ChasmNodeMetadata_CollectionAttributes)(nil),
(*ChasmNodeMetadata_PointerAttributes)(nil),
}
file_temporal_server_api_persistence_v1_chasm_proto_msgTypes[10].OneofWrappers = []any{
(*ChasmNexusCompletion_Success)(nil),
(*ChasmNexusCompletion_Failure)(nil),
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_persistence_v1_chasm_proto_rawDesc), len(file_temporal_server_api_persistence_v1_chasm_proto_rawDesc)),
NumEnums: 0,
NumMessages: 15,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_persistence_v1_chasm_proto_goTypes,
DependencyIndexes: file_temporal_server_api_persistence_v1_chasm_proto_depIdxs,
MessageInfos: file_temporal_server_api_persistence_v1_chasm_proto_msgTypes,
}.Build()
File_temporal_server_api_persistence_v1_chasm_proto = out.File
file_temporal_server_api_persistence_v1_chasm_proto_goTypes = nil
file_temporal_server_api_persistence_v1_chasm_proto_depIdxs = nil
}
func NewTimerSequence(
mutableState historyi.MutableState,
return &timerSequenceImpl{
mutableState: mutableState,
}
}
sequenceIDs := t.LoadAndSortUserTimers()
if len(sequenceIDs) == 0 {
}
firstTimerTask := sequenceIDs[0]
}
sequenceIDs := t.LoadAndSortActivityTimers()
if len(sequenceIDs) == 0 {
}
firstTimerTask := sequenceIDs[0]
}
pendingTimers := t.mutableState.GetPendingTimerInfos()
timers := make(TimerSequenceIDs, 0, len(pendingTimers))
for _, timerInfo := range pendingTimers {
if sequenceID := t.getUserTimerTimeout(
}
return timers
}
// there can be 4 timer per activity
// see TimerType
pendingActivities := t.mutableState.GetPendingActivityInfos()
activityTimers := make(TimerSequenceIDs, 0, len(pendingActivities)*4)
for _, activityInfo := range pendingActivities {
// skip activities that are paused
if activityInfo.Paused {
// Len implements sort.Interface
return len(s)
}
// Swap implements sort.Interface.
}
func init() { file_temporal_server_api_adminservice_v1_request_response_proto_init() }
request_response.pb.go
func file_temporal_server_api_adminservice_v1_request_response_proto_init() {
if File_temporal_server_api_adminservice_v1_request_response_proto != nil {
return
}
file_temporal_server_api_adminservice_v1_request_response_proto_msgTypes[59].OneofWrappers = []any{
(*StreamWorkflowReplicationMessagesRequest_SyncReplicationState)(nil),
}
file_temporal_server_api_adminservice_v1_request_response_proto_msgTypes[60].OneofWrappers = []any{
(*StreamWorkflowReplicationMessagesResponse_Messages)(nil),
}
file_temporal_server_api_adminservice_v1_request_response_proto_msgTypes[61].OneofWrappers = []any{
(*GetNamespaceRequest_Namespace)(nil),
(*GetNamespaceRequest_Id)(nil),
}
file_temporal_server_api_adminservice_v1_request_response_proto_msgTypes[90].OneofWrappers = []any{
(*StartAdminBatchOperationRequest_RefreshTasksOperation)(nil),
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_adminservice_v1_request_response_proto_rawDesc), len(file_temporal_server_api_adminservice_v1_request_response_proto_rawDesc)),
NumEnums: 1,
NumMessages: 105,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_adminservice_v1_request_response_proto_goTypes,
DependencyIndexes: file_temporal_server_api_adminservice_v1_request_response_proto_depIdxs,
EnumInfos: file_temporal_server_api_adminservice_v1_request_response_proto_enumTypes,
MessageInfos: file_temporal_server_api_adminservice_v1_request_response_proto_msgTypes,
}.Build()
File_temporal_server_api_adminservice_v1_request_response_proto = out.File
file_temporal_server_api_adminservice_v1_request_response_proto_goTypes = nil
file_temporal_server_api_adminservice_v1_request_response_proto_depIdxs = nil
}
}
func file_temporal_server_api_replication_v1_message_proto_init() {
if File_temporal_server_api_replication_v1_message_proto != nil {
return
}
file_temporal_server_api_replication_v1_message_proto_msgTypes[0].OneofWrappers = []any{
message.pb.go
(*ReplicationTask_NamespaceTaskAttributes)(nil),
(*ReplicationTask_SyncShardStatusTaskAttributes)(nil),
(*ReplicationTask_SyncActivityTaskAttributes)(nil),
(*ReplicationTask_HistoryTaskAttributes)(nil),
(*ReplicationTask_SyncWorkflowStateTaskAttributes)(nil),
(*ReplicationTask_TaskQueueUserDataAttributes)(nil),
(*ReplicationTask_SyncHsmAttributes)(nil),
(*ReplicationTask_BackfillHistoryTaskAttributes)(nil),
(*ReplicationTask_VerifyVersionedTransitionTaskAttributes)(nil),
(*ReplicationTask_SyncVersionedTransitionTaskAttributes)(nil),
}
file_temporal_server_api_replication_v1_message_proto_msgTypes[21].OneofWrappers = []any{
(*VersionedTransitionArtifact_SyncWorkflowStateMutationAttributes)(nil),
(*VersionedTransitionArtifact_SyncWorkflowStateSnapshotAttributes)(nil),
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_replication_v1_message_proto_rawDesc), len(file_temporal_server_api_replication_v1_message_proto_rawDesc)),
NumEnums: 0,
NumMessages: 23,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_replication_v1_message_proto_goTypes,
DependencyIndexes: file_temporal_server_api_replication_v1_message_proto_depIdxs,
MessageInfos: file_temporal_server_api_replication_v1_message_proto_msgTypes,
}.Build()
File_temporal_server_api_replication_v1_message_proto = out.File
file_temporal_server_api_replication_v1_message_proto_goTypes = nil
file_temporal_server_api_replication_v1_message_proto_depIdxs = nil
}
resolver ReplicationResolver,
mutations ...Mutation,
if resolver == nil {
return nil, serviceerror.NewInvalidArgument("replicationResolver must be provided")
}
info: detail.Info,
config: detail.Config,
configVersion: detail.ConfigVersion,
customSearchAttributesMapper: CustomSearchAttributesMapper{
fieldToAlias: detail.Config.CustomSearchAttributeAliases,
aliasToField: util.InverseMap(detail.Config.CustomSearchAttributeAliases),
},
replicationResolver: resolver,
}
for _, m := range mutations {
}
}
// ID observes this namespace's permanent unique identifier in string form.
if ns.info == nil {
return ID("")
}
}
// Name observes this namespace's configured name.
if ns.info == nil {
return Name("")
}
}
// FailoverVersion return the namespace failover version
return ns.replicationResolver.FailoverVersion(businessID)
}
// IsGlobalNamespace returns whether the namespace is a global namespace.
// Retention returns retention duration for this namespace.
if ns.config.Retention == nil {
}
return ns.config.Retention.AsDuration()
}
return string(id)
}
func (id ID) IsEmpty() bool {
}
return string(n)
}
func (n Name) IsEmpty() bool {
config *persistencespb.NamespaceConfig,
targetCluster string,
detail := &persistencespb.NamespaceDetail{
Info: ensureInfo(info),
Config: ensureConfig(config),
ReplicationConfig: &persistencespb.NamespaceReplicationConfig{
ActiveClusterName: targetCluster,
Clusters: []string{targetCluster},
},
FailoverVersion: common.EmptyVersion,
}
factory := NewDefaultReplicationResolverFactory()
resolver := factory(detail)
ns, _ := FromPersistentState(detail, resolver, WithGlobalFlag(false))
return ns
}
// NewNamespaceForTest returns an entry with test data
repConfig *persistencespb.NamespaceReplicationConfig,
failoverVersion int64,
detail := &persistencespb.NamespaceDetail{
Info: ensureInfo(info),
Config: ensureConfig(config),
ReplicationConfig: ensureRepConfig(repConfig),
FailoverVersion: failoverVersion,
}
factory := NewDefaultReplicationResolverFactory()
resolver := factory(detail)
ns, _ := FromPersistentState(detail, resolver, WithGlobalFlag(true))
return ns
}
func ensureInfo(proto *persistencespb.NamespaceInfo) *persistencespb.NamespaceInfo {
testconstructors.go
if proto == nil {
return &persistencespb.NamespaceInfo{}
}
}
func ensureConfig(proto *persistencespb.NamespaceConfig) *persistencespb.NamespaceConfig {
testconstructors.go
if proto == nil {
return &persistencespb.NamespaceConfig{}
}
}
func ensureRepConfig(proto *persistencespb.NamespaceReplicationConfig) *persistencespb.NamespaceReplicationConfig {
testconstructors.go
if proto == nil {
return &persistencespb.NamespaceReplicationConfig{}
}
}
func (*UpdateInfo) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_update_proto_msgTypes[3]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
}
func file_temporal_server_api_persistence_v1_update_proto_init() {
if File_temporal_server_api_persistence_v1_update_proto != nil {
return
}
file_temporal_server_api_persistence_v1_hsm_proto_init()
file_temporal_server_api_persistence_v1_update_proto_msgTypes[0].OneofWrappers = []any{
(*UpdateAdmissionInfo_HistoryPointer_)(nil),
}
file_temporal_server_api_persistence_v1_update_proto_msgTypes[3].OneofWrappers = []any{
(*UpdateInfo_Acceptance)(nil),
(*UpdateInfo_Completion)(nil),
(*UpdateInfo_Admission)(nil),
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_persistence_v1_update_proto_rawDesc), len(file_temporal_server_api_persistence_v1_update_proto_rawDesc)),
NumEnums: 0,
NumMessages: 5,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_persistence_v1_update_proto_goTypes,
DependencyIndexes: file_temporal_server_api_persistence_v1_update_proto_depIdxs,
MessageInfos: file_temporal_server_api_persistence_v1_update_proto_msgTypes,
}.Build()
File_temporal_server_api_persistence_v1_update_proto = out.File
file_temporal_server_api_persistence_v1_update_proto_goTypes = nil
file_temporal_server_api_persistence_v1_update_proto_depIdxs = nil
}
logger log.Logger,
renewRangeIDFn renewRangeIDFn,
return &taskKeyGenerator{
nextTaskID: taskIDUninitialized,
exclusiveMaxTaskID: taskIDUninitialized,
rangeSizeBits: rangeSizeBits,
timeSource: timeSource,
logger: logger,
renewRangeIDFn: renewRangeIDFn,
}
}
func (a *taskKeyGenerator) setTaskKeys(
func (a *taskKeyGenerator) peekTaskKey(
category tasks.Category,
switch category.Type() {
return tasks.NewImmediateKey(a.nextTaskID)
case tasks.CategoryTypeScheduled:
return tasks.NewKey(
func (a *taskKeyGenerator) generateTaskKey(
category tasks.Category,
id, err := a.generateTaskID()
if err != nil {
return tasks.Key{}, err
}
case tasks.CategoryTypeImmediate:
return tasks.NewImmediateKey(id), nil
case tasks.CategoryTypeScheduled:
return tasks.NewKey(
}
a.nextTaskID = rangeID << a.rangeSizeBits
a.exclusiveMaxTaskID = (rangeID + 1) << a.rangeSizeBits
a.logger.Info("Task key range updated",
tag.Number(a.nextTaskID),
tag.NextNumber(a.exclusiveMaxTaskID),
)
}
func (a *taskKeyGenerator) setTaskMinScheduledTime(
}
if a.nextTaskID == taskIDUninitialized {
a.logger.Panic("Range id is not initialized before generating task id")
}
if err := a.renewRangeIDFn(); err != nil {
return taskIDUninitialized, err
}
a.nextTaskID++
return taskID, nil
}
// NewExponentialRetryPolicy returns an instance of ExponentialRetryPolicy using the provided initialInterval
func NewExponentialRetryPolicy(initialInterval time.Duration) *ExponentialRetryPolicy {
retrypolicy.go
p := &ExponentialRetryPolicy{
initialInterval: initialInterval,
backoffCoefficient: defaultBackoffCoefficient,
maximumInterval: defaultMaximumInterval,
expirationInterval: defaultExpirationInterval,
maximumAttempts: defaultMaximumAttempts,
}
return p
}
// NewRetrier is used for creating a new instance of Retrier
// This does *not* cause the policy to stop retrying when the interval between retries reaches the supplied duration.
// That is what WithExpirationInterval does. Instead, this prevents the interval from exceeding maximumInterval.
func (p *ExponentialRetryPolicy) WithMaximumInterval(maximumInterval time.Duration) *ExponentialRetryPolicy {
retrypolicy.go
p.maximumInterval = maximumInterval
return p
}
// WithExpirationInterval sets the absolute expiration interval for all retries
func (p *ExponentialRetryPolicy) WithExpirationInterval(expirationInterval time.Duration) *ExponentialRetryPolicy {
retrypolicy.go
p.expirationInterval = expirationInterval
return p
}
// WithMaximumAttempts sets the maximum number of retry attempts
func (p *ExponentialRetryPolicy) WithMaximumAttempts(maximumAttempts int) *ExponentialRetryPolicy {
retrypolicy.go
p.maximumAttempts = maximumAttempts
return p
}
// ComputeNextDelay returns the next delay interval. This is used by Retrier to delay calling the operation again
var _ RetryPolicy = (*ConstantDelayRetryPolicy)(nil)
func NewConstantDelayRetryPolicy(delay time.Duration) *ConstantDelayRetryPolicy {
retrypolicy.go
return &ConstantDelayRetryPolicy{
maximumAttempts: defaultMaximumAttempts,
jitterPct: defaultJitterPct,
delay: delay,
}
}
func (p *ConstantDelayRetryPolicy) WithMaximumAttempts(maximumAttempts int) *ConstantDelayRetryPolicy {
retrypolicy.go
p.maximumAttempts = maximumAttempts
return p
}
func (p *ConstantDelayRetryPolicy) WithJitter(jitterPct float64) *ConstantDelayRetryPolicy {
// NewVersionHistory create a new instance of VersionHistory.
func NewVersionHistory(branchToken []byte, items []*historyspb.VersionHistoryItem) *historyspb.VersionHistory {
version_history.go
return &historyspb.VersionHistory{
BranchToken: branchToken,
Items: items,
}
}
// CopyVersionHistory copies VersionHistory.
// SetVersionHistoryBranchToken sets the branch token.
func SetVersionHistoryBranchToken(v *historyspb.VersionHistory, branchToken []byte) {
version_history.go
v.BranchToken = make([]byte, len(branchToken))
copy(v.BranchToken, branchToken)
}
// AddOrUpdateVersionHistoryItem updates the VersionHistory with new VersionHistoryItem.
func AddOrUpdateVersionHistoryItem(v *historyspb.VersionHistory, item *historyspb.VersionHistoryItem) error {
version_history.go
if len(v.Items) == 0 {
return nil
}
if item.Version < lastItem.Version {
return serviceerror.NewInternalf("cannot update version history with a lower version %v. Last version: %v", item.Version, lastItem.Version)
}
return serviceerror.NewInternalf("cannot add version history with a lower event id %v. Last event id: %v", item.GetEventId(), lastItem.GetEventId())
}
// Add a new history
v.Items = append(v.Items, CopyVersionHistoryItem(item))
// Update event ID
lastItem.EventId = item.GetEventId()
}
}
// GetLastVersionHistoryItem return the last VersionHistoryItem.
func GetLastVersionHistoryItem(v *historyspb.VersionHistory) (*historyspb.VersionHistoryItem, error) {
version_history.go
return getLastVersionHistoryItem(v.Items)
}
func getLastVersionHistoryItem(v []*historyspb.VersionHistoryItem) (*historyspb.VersionHistoryItem, error) {
version_history.go
if len(v) == 0 {
return nil, serviceerror.NewInternal("version history is empty.")
}
}
// IsEmptyVersionHistory indicate whether version history is empty
return len(v.Items) == 0
}
// CompareVersionHistory compares 2 version history items
// New creates a new cache with the given options
return NewWithMetrics(maxSize, opts, metrics.NoopMetricsHandler)
}
// NewWithMetrics creates a new cache that will emit capacity and ttl metrics.
// handler should be tagged with metrics.CacheTypeTag.
func NewWithMetrics(maxSize int, opts *Options, handler metrics.Handler) StoppableCache {
lru.go
if opts == nil {
opts = &Options{}
}
if backgroundEvict == nil {
return dynamicconfig.CacheBackgroundEvictSettings{
Enabled: false,
}
}
}
if timeSource == nil {
}
metrics.CacheTtl.With(handler).Record(opts.TTL)
c := &lru{
byAccess: list.New(),
byKey: make(map[any]*list.Element),
ttl: opts.TTL,
maxSize: maxSize,
currSize: 0,
pin: opts.Pin,
onPut: opts.OnPut,
onEvict: opts.OnEvict,
timeSource: timeSource,
metricsHandler: handler,
backgroundEvict: backgroundEvict,
}
if c.backgroundEvict().Enabled {
c.loops.Go(c.bgEvictLoop)
}
}
// WorkflowAction returns tag for WorkflowAction
return NewStringTag("wf-action", action)
}
// WorkflowListFilterType returns tag for WorkflowListFilterType
return NewStringTag("wf-list-filter-type", listFilterType)
}
// general
// Component returns tag for Component
return NewStringTag("component", component)
}
// Lifecycle returns tag for Lifecycle
return NewStringTag("lifecycle", lifecycle)
}
// StoreOperation returns tag for StoreOperation
return NewStringTag("store-operation", storeOperation)
}
// OperationResult returns tag for OperationResult
return NewStringTag("operation-result", operationResult)
}
// ErrorType returns tag for ErrorType
// errorType returns tag for ErrorType given a string
return NewStringTag("error-type", errorType)
}
// Shardupdate returns tag for Shardupdate
return NewStringTag("shard-update", shardupdate)
}
// scope returns a tag for scope
// Pre-defined scope tags are in values.go.
return NewStringTag("scope", scope)
}
// general
// Number returns tag for Number
return NewInt64("number", n)
}
// NextNumber returns tag for NextNumber
return NewInt64("next-number", n)
}
// ServerName returns tag for ServerName
func (*HistoryBranch) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_history_tree_proto_msgTypes[1]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
func (*HistoryBranchRange) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_history_tree_proto_msgTypes[2]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
}
func init() { file_temporal_server_api_persistence_v1_history_tree_proto_init() }
history_tree.pb.go
func file_temporal_server_api_persistence_v1_history_tree_proto_init() {
if File_temporal_server_api_persistence_v1_history_tree_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_persistence_v1_history_tree_proto_rawDesc), len(file_temporal_server_api_persistence_v1_history_tree_proto_rawDesc)),
NumEnums: 0,
NumMessages: 3,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_persistence_v1_history_tree_proto_goTypes,
DependencyIndexes: file_temporal_server_api_persistence_v1_history_tree_proto_depIdxs,
MessageInfos: file_temporal_server_api_persistence_v1_history_tree_proto_msgTypes,
}.Build()
File_temporal_server_api_persistence_v1_history_tree_proto = out.File
file_temporal_server_api_persistence_v1_history_tree_proto_goTypes = nil
file_temporal_server_api_persistence_v1_history_tree_proto_depIdxs = nil
}
// deepCopyForMapstructure does a simple deep copy of T. Fancy cases (anything other than plain old data)
// is not handled and will panic.
// nolint:revive // this will be triggered from a static initializer before it can be triggered from production code
return deepCopyValue(reflect.ValueOf(t)).Interface().(T)
}
switch v.Kind() {
case reflect.Bool, reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64,
reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64,
nv := reflect.New(v.Type()).Elem()
nv.Set(v)
return nv
case reflect.Array:
nv := reflect.New(v.Type()).Elem()
}
return deepCopyValue(v.Elem()).Addr()
if v.IsNil() {
return v
}
nv := reflect.MakeSlice(v.Type(), v.Len(), v.Len())
for i := range v.Len() {
}
return nv
// Special case for time.Time: it has unexported fields so we can't copy it field by
// field, but we can copy zero values (which is all we need for default values).
if v.Type() == reflect.TypeFor[time.Time]() {
if v.Interface().(time.Time).IsZero() {
return reflect.ValueOf(time.Time{})
}
// nolint:forbidigo // this will be triggered from a static initializer before it can be triggered from production code
panic(fmt.Sprintf("Can't deep copy non-zero time.Time: %v", v.Interface()))
}
for i := range v.Type().NumField() {
nv.Field(i).Set(deepCopyValue(v.Field(i)))
}
return nv
case reflect.Interface, reflect.Func, reflect.Chan:
// only nil values of any other reference types allowed!
if v.IsNil() {
return v
}
fallthrough
default:
)
b := make([]string, len(a))
for i, v := range a {
b[i] = f(v)
}
return b
}
return fmt.Sprintf(deleteMapQryTemplate, tableName)
}
func makeSetKeyInMapQry(tableName string, nonPrimaryKeyColumns []string, mapKeyName string) string {
execution_maps.go
return fmt.Sprintf(setKeyInMapQryTemplate,
tableName,
strings.Join(nonPrimaryKeyColumns, ","),
strings.Join(stringMap(nonPrimaryKeyColumns, func(x string) string {
return ":" + x
}), ","),
strings.Join(stringMap(nonPrimaryKeyColumns, func(x string) string {
return x + "=VALUES(" + x + ")"
}), ","),
mapKeyName)
}
return fmt.Sprintf(deleteKeyInMapQryTemplate,
tableName,
mapKeyName)
}
func makeGetMapQryTemplate(tableName string, nonPrimaryKeyColumns []string, mapKeyName string) string {
execution_maps.go
return fmt.Sprintf(getMapQryTemplate,
tableName,
mapKeyName,
strings.Join(nonPrimaryKeyColumns, ","))
}
var (
)
b := make([]string, len(a))
for i, v := range a {
b[i] = f(v)
}
return b
}
return fmt.Sprintf(deleteMapQueryTemplate, tableName)
}
func makeSetKeyInMapQry(tableName string, nonPrimaryKeyColumns []string, mapKeyName string) string {
execution_maps.go
return fmt.Sprintf(setKeyInMapQueryTemplate,
tableName,
strings.Join(nonPrimaryKeyColumns, ","),
strings.Join(stringMap(nonPrimaryKeyColumns, func(x string) string {
return ":" + x
}), ","),
mapKeyName,
return "excluded." + x
}), ","))
}
return fmt.Sprintf(deleteKeyInMapQueryTemplate,
tableName,
mapKeyName)
}
func makeGetMapQryTemplate(tableName string, nonPrimaryKeyColumns []string, mapKeyName string) string {
execution_maps.go
return fmt.Sprintf(getMapQueryTemplate,
tableName,
mapKeyName,
strings.Join(nonPrimaryKeyColumns, ","))
}
var (
)
b := make([]string, len(a))
for i, v := range a {
b[i] = f(v)
}
return b
}
return fmt.Sprintf(deleteMapQryTemplate, tableName)
}
func makeSetKeyInMapQry(tableName string, nonPrimaryKeyColumns []string, mapKeyName string) string {
execution_maps.go
return fmt.Sprintf(setKeyInMapQryTemplate,
tableName,
strings.Join(nonPrimaryKeyColumns, ","),
strings.Join(stringMap(nonPrimaryKeyColumns, func(x string) string {
return ":" + x
}), ","),
strings.Join(stringMap(nonPrimaryKeyColumns, func(x string) string {
return x + "=" + x
}), ","),
mapKeyName)
}
return fmt.Sprintf(deleteKeyInMapQryTemplate,
tableName,
mapKeyName)
}
func makeGetMapQryTemplate(tableName string, nonPrimaryKeyColumns []string, mapKeyName string) string {
execution_maps.go
return fmt.Sprintf(getMapQryTemplate,
tableName,
mapKeyName,
strings.Join(nonPrimaryKeyColumns, ","))
}
var (
}
dbCustomSearchAttributeFieldNameRE = func() map[enumspb.IndexedValueType]*regexp.Regexp {
constants.go
res := map[enumspb.IndexedValueType]*regexp.Regexp{}
for t := range defaultNumDBCustomSearchAttributes {
res[t] = regexp.MustCompile(fmt.Sprintf(`^%s(0[1-9]|[1-9][0-9])$`, t.String()))
}
return res
}()
)
// System returns a clone of the system search attributes map.
return maps.Clone(system)
}
// Predefined returns a clone of the predefined search attributes map.
return maps.Clone(predefined)
}
// PredefinedWhiteList returns a clone of the predefined whitelist search attributes map.
return maps.Clone(predefinedWhiteList)
}
// Reserved returns a clone of the reserved field names map.
// GetSqlDbColName maps system and reserved search attributes to column names for SQL tables.
// If the input is not a system or reserved search attribute, then it returns the input.
if fieldName, ok := sqlDbSystemNameToColName[name]; ok {
}
return name
}
func GetDBIndexSearchAttributes(
override map[enumspb.IndexedValueType]int,
csa := map[string]enumspb.IndexedValueType{}
for saType, defaultNumAttrs := range defaultNumDBCustomSearchAttributes {
numAttrs := defaultNumAttrs
if value, ok := override[saType]; ok {
numAttrs = value
}
csa[fmt.Sprintf("%s%02d", saType.String(), i+1)] = saType
}
}
CustomSearchAttributes: csa,
}
}
}
func file_temporal_server_api_taskqueue_v1_message_proto_init() {
if File_temporal_server_api_taskqueue_v1_message_proto != nil {
return
}
file_temporal_server_api_taskqueue_v1_message_proto_msgTypes[0].OneofWrappers = []any{
message.pb.go
(*TaskVersionDirective_UseAssignmentRules)(nil),
(*TaskVersionDirective_AssignedBuildId)(nil),
}
file_temporal_server_api_taskqueue_v1_message_proto_msgTypes[5].OneofWrappers = []any{
(*TaskQueuePartition_NormalPartitionId)(nil),
(*TaskQueuePartition_StickyName)(nil),
(*TaskQueuePartition_WorkerCommands)(nil),
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_taskqueue_v1_message_proto_rawDesc), len(file_temporal_server_api_taskqueue_v1_message_proto_rawDesc)),
NumEnums: 0,
NumMessages: 16,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_taskqueue_v1_message_proto_goTypes,
DependencyIndexes: file_temporal_server_api_taskqueue_v1_message_proto_depIdxs,
MessageInfos: file_temporal_server_api_taskqueue_v1_message_proto_msgTypes,
}.Build()
File_temporal_server_api_taskqueue_v1_message_proto = out.File
file_temporal_server_api_taskqueue_v1_message_proto_goTypes = nil
file_temporal_server_api_taskqueue_v1_message_proto_depIdxs = nil
}
func (*VectorClock) ProtoMessage() {}
mi := &file_temporal_server_api_clock_v1_message_proto_msgTypes[0]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
}
if x != nil {
return x.Clock
}
return 0
}
}
func file_temporal_server_api_clock_v1_message_proto_init() {
if File_temporal_server_api_clock_v1_message_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_clock_v1_message_proto_rawDesc), len(file_temporal_server_api_clock_v1_message_proto_rawDesc)),
NumEnums: 0,
NumMessages: 2,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_clock_v1_message_proto_goTypes,
DependencyIndexes: file_temporal_server_api_clock_v1_message_proto_depIdxs,
MessageInfos: file_temporal_server_api_clock_v1_message_proto_msgTypes,
}.Build()
File_temporal_server_api_clock_v1_message_proto = out.File
file_temporal_server_api_clock_v1_message_proto_goTypes = nil
file_temporal_server_api_clock_v1_message_proto_depIdxs = nil
}
// NewMockBean creates a new mock instance.
mock := &MockBean{ctrl: ctrl}
mock.recorder = &MockBeanMockRecorder{mock}
return mock
}
// EXPECT returns an object that allows the caller to indicate expected use.
return m.recorder
}
// Close mocks base method.
// GetFrontendClient indicates an expected call of GetFrontendClient.
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetFrontendClient", reflect.TypeOf((*MockBean)(nil).GetFrontendClient))
}
// GetHistoryClient mocks base method.
// GetHistoryClient indicates an expected call of GetHistoryClient.
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetHistoryClient", reflect.TypeOf((*MockBean)(nil).GetHistoryClient))
}
// GetMatchingClient mocks base method.
// GetMatchingClient indicates an expected call of GetMatchingClient.
func (mr *MockBeanMockRecorder) GetMatchingClient(namespaceIDToName any) *gomock.Call {
client_bean_mock.go
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetMatchingClient", reflect.TypeOf((*MockBean)(nil).GetMatchingClient), namespaceIDToName)
}
// GetRemoteAdminClient mocks base method.
// GetRemoteAdminClient indicates an expected call of GetRemoteAdminClient.
func (mr *MockBeanMockRecorder) GetRemoteAdminClient(arg0 any) *gomock.Call {
client_bean_mock.go
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetRemoteAdminClient", reflect.TypeOf((*MockBean)(nil).GetRemoteAdminClient), arg0)
}
// GetRemoteFrontendClient mocks base method.
// GetRemoteFrontendClient indicates an expected call of GetRemoteFrontendClient.
func (mr *MockBeanMockRecorder) GetRemoteFrontendClient(arg0 any) *gomock.Call {
client_bean_mock.go
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetRemoteFrontendClient", reflect.TypeOf((*MockBean)(nil).GetRemoteFrontendClient), arg0)
}
func (*QueueState) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_queues_proto_msgTypes[1]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
}
func file_temporal_server_api_persistence_v1_queues_proto_init() {
if File_temporal_server_api_persistence_v1_queues_proto != nil {
return
}
file_temporal_server_api_persistence_v1_predicates_proto_init()
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_persistence_v1_queues_proto_rawDesc), len(file_temporal_server_api_persistence_v1_queues_proto_rawDesc)),
NumEnums: 0,
NumMessages: 12,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_persistence_v1_queues_proto_goTypes,
DependencyIndexes: file_temporal_server_api_persistence_v1_queues_proto_depIdxs,
MessageInfos: file_temporal_server_api_persistence_v1_queues_proto_msgTypes,
}.Build()
File_temporal_server_api_persistence_v1_queues_proto = out.File
file_temporal_server_api_persistence_v1_queues_proto_goTypes = nil
file_temporal_server_api_persistence_v1_queues_proto_depIdxs = nil
}
logger log.Logger,
renewRangeIDFn renewRangeIDFn,
return &taskKeyManager{
generator: newTaskKeyGenerator(
config.RangeSizeBits,
timeSource,
logger,
renewRangeIDFn,
),
tracker: newTaskRequestTracker(taskCategoryRegistry),
timeSource: timeSource,
logger: logger,
config: config,
}
}
func (m *taskKeyManager) setAndTrackTaskKeys(
func (m *taskKeyManager) peekTaskKey(
category tasks.Category,
return m.generator.peekTaskKey(category)
}
func (m *taskKeyManager) generateTaskKey(
category tasks.Category,
return m.generator.generateTaskKey(category)
}
func (m *taskKeyManager) drainTaskRequests() {
func (m *taskKeyManager) setRangeID(
rangeID int64,
m.generator.setRangeID(rangeID)
// rangeID update means all pending add tasks requests either already succeeded
// are guaranteed to fail, so we can clear pending requests in the tracker
m.tracker.clear()
}
func (m *taskKeyManager) setTaskMinScheduledTime(
}
return file_temporal_server_api_enums_v1_common_proto_enumTypes[1].Descriptor()
}
func (ChecksumFlavor) Type() protoreflect.EnumType {
}
return file_temporal_server_api_enums_v1_common_proto_enumTypes[2].Descriptor()
}
func (CallbackState) Type() protoreflect.EnumType {
}
func file_temporal_server_api_enums_v1_common_proto_init() {
if File_temporal_server_api_enums_v1_common_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_enums_v1_common_proto_rawDesc), len(file_temporal_server_api_enums_v1_common_proto_rawDesc)),
NumEnums: 3,
NumMessages: 0,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_enums_v1_common_proto_goTypes,
DependencyIndexes: file_temporal_server_api_enums_v1_common_proto_depIdxs,
EnumInfos: file_temporal_server_api_enums_v1_common_proto_enumTypes,
}.Build()
File_temporal_server_api_enums_v1_common_proto = out.File
file_temporal_server_api_enums_v1_common_proto_goTypes = nil
file_temporal_server_api_enums_v1_common_proto_depIdxs = nil
}
}
return file_temporal_server_api_enums_v1_task_proto_enumTypes[1].Descriptor()
}
func (TaskType) Type() protoreflect.EnumType {
}
return file_temporal_server_api_enums_v1_task_proto_enumTypes[2].Descriptor()
}
func (TaskPriority) Type() protoreflect.EnumType {
}
func file_temporal_server_api_enums_v1_task_proto_init() {
if File_temporal_server_api_enums_v1_task_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_enums_v1_task_proto_rawDesc), len(file_temporal_server_api_enums_v1_task_proto_rawDesc)),
NumEnums: 3,
NumMessages: 0,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_enums_v1_task_proto_goTypes,
DependencyIndexes: file_temporal_server_api_enums_v1_task_proto_depIdxs,
EnumInfos: file_temporal_server_api_enums_v1_task_proto_enumTypes,
}.Build()
File_temporal_server_api_enums_v1_task_proto = out.File
file_temporal_server_api_enums_v1_task_proto_goTypes = nil
file_temporal_server_api_enums_v1_task_proto_depIdxs = nil
}
}
return file_temporal_server_api_enums_v1_workflow_proto_enumTypes[0].Descriptor()
}
func (WorkflowExecutionState) Type() protoreflect.EnumType {
}
return file_temporal_server_api_enums_v1_workflow_proto_enumTypes[1].Descriptor()
}
func (WorkflowBackoffType) Type() protoreflect.EnumType {
}
func file_temporal_server_api_enums_v1_workflow_proto_init() {
if File_temporal_server_api_enums_v1_workflow_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_enums_v1_workflow_proto_rawDesc), len(file_temporal_server_api_enums_v1_workflow_proto_rawDesc)),
NumEnums: 3,
NumMessages: 0,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_enums_v1_workflow_proto_goTypes,
DependencyIndexes: file_temporal_server_api_enums_v1_workflow_proto_depIdxs,
EnumInfos: file_temporal_server_api_enums_v1_workflow_proto_enumTypes,
}.Build()
File_temporal_server_api_enums_v1_workflow_proto = out.File
file_temporal_server_api_enums_v1_workflow_proto_goTypes = nil
file_temporal_server_api_enums_v1_workflow_proto_depIdxs = nil
}
// NewMockMonitor creates a new mock instance.
mock := &MockMonitor{ctrl: ctrl}
mock.recorder = &MockMonitorMockRecorder{mock}
return mock
}
// EXPECT returns an object that allows the caller to indicate expected use.
return m.recorder
}
// ApproximateMaxPropagationTime mocks base method.
// GetResolver indicates an expected call of GetResolver.
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetResolver", reflect.TypeOf((*MockMonitor)(nil).GetResolver), service)
}
// SetDraining mocks base method.
// WaitUntilInitialized indicates an expected call of WaitUntilInitialized.
func (mr *MockMonitorMockRecorder) WaitUntilInitialized(arg0 any) *gomock.Call {
interfaces_mock.go
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "WaitUntilInitialized", reflect.TypeOf((*MockMonitor)(nil).WaitUntilInitialized), arg0)
}
// MockServiceResolver is a mock of ServiceResolver interface.
// NewMockServiceResolver creates a new mock instance.
func NewMockServiceResolver(ctrl *gomock.Controller) *MockServiceResolver {
interfaces_mock.go
mock := &MockServiceResolver{ctrl: ctrl}
mock.recorder = &MockServiceResolverMockRecorder{mock}
return mock
}
// EXPECT returns an object that allows the caller to indicate expected use.
// NewMockHostInfoProvider creates a new mock instance.
func NewMockHostInfoProvider(ctrl *gomock.Controller) *MockHostInfoProvider {
interfaces_mock.go
mock := &MockHostInfoProvider{ctrl: ctrl}
mock.recorder = &MockHostInfoProviderMockRecorder{mock}
return mock
}
// EXPECT returns an object that allows the caller to indicate expected use.
}
func init() { file_temporal_server_chasm_lib_activity_proto_v1_activity_state_proto_init() }
activity_state.pb.go
func file_temporal_server_chasm_lib_activity_proto_v1_activity_state_proto_init() {
if File_temporal_server_chasm_lib_activity_proto_v1_activity_state_proto != nil {
return
}
file_temporal_server_chasm_lib_activity_proto_v1_activity_state_proto_msgTypes[7].OneofWrappers = []any{
activity_state.pb.go
(*ActivityOutcome_Successful_)(nil),
(*ActivityOutcome_Failed_)(nil),
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_chasm_lib_activity_proto_v1_activity_state_proto_rawDesc), len(file_temporal_server_chasm_lib_activity_proto_v1_activity_state_proto_rawDesc)),
NumEnums: 2,
NumMessages: 11,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_chasm_lib_activity_proto_v1_activity_state_proto_goTypes,
DependencyIndexes: file_temporal_server_chasm_lib_activity_proto_v1_activity_state_proto_depIdxs,
EnumInfos: file_temporal_server_chasm_lib_activity_proto_v1_activity_state_proto_enumTypes,
MessageInfos: file_temporal_server_chasm_lib_activity_proto_v1_activity_state_proto_msgTypes,
}.Build()
File_temporal_server_chasm_lib_activity_proto_v1_activity_state_proto = out.File
file_temporal_server_chasm_lib_activity_proto_v1_activity_state_proto_goTypes = nil
file_temporal_server_chasm_lib_activity_proto_v1_activity_state_proto_depIdxs = nil
}
}
func init() { file_temporal_server_chasm_lib_nexusoperation_proto_v1_operation_proto_init() }
operation.pb.go
func file_temporal_server_chasm_lib_nexusoperation_proto_v1_operation_proto_init() {
if File_temporal_server_chasm_lib_nexusoperation_proto_v1_operation_proto != nil {
return
}
file_temporal_server_chasm_lib_nexusoperation_proto_v1_operation_proto_msgTypes[2].OneofWrappers = []any{
operation.pb.go
(*OperationOutcome_Successful_)(nil),
(*OperationOutcome_Failed_)(nil),
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_chasm_lib_nexusoperation_proto_v1_operation_proto_rawDesc), len(file_temporal_server_chasm_lib_nexusoperation_proto_v1_operation_proto_rawDesc)),
NumEnums: 2,
NumMessages: 8,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_chasm_lib_nexusoperation_proto_v1_operation_proto_goTypes,
DependencyIndexes: file_temporal_server_chasm_lib_nexusoperation_proto_v1_operation_proto_depIdxs,
EnumInfos: file_temporal_server_chasm_lib_nexusoperation_proto_v1_operation_proto_enumTypes,
MessageInfos: file_temporal_server_chasm_lib_nexusoperation_proto_v1_operation_proto_msgTypes,
}.Build()
File_temporal_server_chasm_lib_nexusoperation_proto_v1_operation_proto = out.File
file_temporal_server_chasm_lib_nexusoperation_proto_v1_operation_proto_goTypes = nil
file_temporal_server_chasm_lib_nexusoperation_proto_v1_operation_proto_depIdxs = nil
}
// NewRoute returns a new [Route] instance with the given components.
return Route[T]{components: components}
}
// RouteBuilder is a builder for the [Route] interface.
// NewBuilder creates a new [RouteBuilder] instance, which can be used to define a new [Route] via a fluent API.
return &RouteBuilder[T]{}
}
// With adds a series of [Component] instances to the [Route].
r.components = append(r.components, c...)
return r
}
// Constant adds a [Constant] component to the [Route].
return r.With(Constant[T](values...))
}
// StringVariable adds a [StringVariable] component to the [Route].
func (r *RouteBuilder[T]) StringVariable(name string, getter func(*T) *string) *RouteBuilder[T] {
route.go
return r.With(StringVariable[T](name, getter))
}
// Build returns a read-only [Route].
return NewRoute[T](r.components...)
}
// Representation returns the [github.com/gorilla/mux] compatible string representation of the route for usage in a
// Constant returns a [Component] that represents a series of constant HTTP path components in a Route.
// They will be joined via strings when used to construct a path or path representation.
return values
}
type constant[T any] []string
// StringVariable returns a [Component] that represents a string variable in a Route.
return stringVariable[T]{name, getter}
}
type stringVariable[T any] struct {
}
func file_temporal_server_api_persistence_v1_nexus_proto_init() {
if File_temporal_server_api_persistence_v1_nexus_proto != nil {
return
}
file_temporal_server_api_persistence_v1_nexus_proto_msgTypes[1].OneofWrappers = []any{
nexus.pb.go
(*NexusEndpointTarget_Worker_)(nil),
(*NexusEndpointTarget_External_)(nil),
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_persistence_v1_nexus_proto_rawDesc), len(file_temporal_server_api_persistence_v1_nexus_proto_rawDesc)),
NumEnums: 0,
NumMessages: 6,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_persistence_v1_nexus_proto_goTypes,
DependencyIndexes: file_temporal_server_api_persistence_v1_nexus_proto_depIdxs,
MessageInfos: file_temporal_server_api_persistence_v1_nexus_proto_msgTypes,
}.Build()
File_temporal_server_api_persistence_v1_nexus_proto = out.File
file_temporal_server_api_persistence_v1_nexus_proto_goTypes = nil
file_temporal_server_api_persistence_v1_nexus_proto_depIdxs = nil
}
}
func init() { file_temporal_server_api_persistence_v1_workflow_mutable_state_proto_init() }
workflow_mutable_state.pb.go
func file_temporal_server_api_persistence_v1_workflow_mutable_state_proto_init() {
if File_temporal_server_api_persistence_v1_workflow_mutable_state_proto != nil {
return
}
file_temporal_server_api_persistence_v1_executions_proto_init()
file_temporal_server_api_persistence_v1_hsm_proto_init()
file_temporal_server_api_persistence_v1_update_proto_init()
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_persistence_v1_workflow_mutable_state_proto_rawDesc), len(file_temporal_server_api_persistence_v1_workflow_mutable_state_proto_rawDesc)),
NumEnums: 0,
NumMessages: 16,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_persistence_v1_workflow_mutable_state_proto_goTypes,
DependencyIndexes: file_temporal_server_api_persistence_v1_workflow_mutable_state_proto_depIdxs,
MessageInfos: file_temporal_server_api_persistence_v1_workflow_mutable_state_proto_msgTypes,
}.Build()
File_temporal_server_api_persistence_v1_workflow_mutable_state_proto = out.File
file_temporal_server_api_persistence_v1_workflow_mutable_state_proto_goTypes = nil
file_temporal_server_api_persistence_v1_workflow_mutable_state_proto_depIdxs = nil
}
func (*BaseExecutionInfo) ProtoMessage() {}
mi := &file_temporal_server_api_workflow_v1_message_proto_msgTypes[2]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
}
func file_temporal_server_api_workflow_v1_message_proto_init() {
if File_temporal_server_api_workflow_v1_message_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_workflow_v1_message_proto_rawDesc), len(file_temporal_server_api_workflow_v1_message_proto_rawDesc)),
NumEnums: 0,
NumMessages: 3,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_workflow_v1_message_proto_goTypes,
DependencyIndexes: file_temporal_server_api_workflow_v1_message_proto_depIdxs,
MessageInfos: file_temporal_server_api_workflow_v1_message_proto_msgTypes,
}.Build()
File_temporal_server_api_workflow_v1_message_proto = out.File
file_temporal_server_api_workflow_v1_message_proto_goTypes = nil
file_temporal_server_api_workflow_v1_message_proto_depIdxs = nil
}
}
func init() { file_temporal_server_chasm_lib_callback_proto_v1_message_proto_init() }
message.pb.go
func file_temporal_server_chasm_lib_callback_proto_v1_message_proto_init() {
if File_temporal_server_chasm_lib_callback_proto_v1_message_proto != nil {
return
}
file_temporal_server_chasm_lib_callback_proto_v1_message_proto_msgTypes[1].OneofWrappers = []any{
message.pb.go
(*Callback_Nexus_)(nil),
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_chasm_lib_callback_proto_v1_message_proto_rawDesc), len(file_temporal_server_chasm_lib_callback_proto_v1_message_proto_rawDesc)),
NumEnums: 1,
NumMessages: 5,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_chasm_lib_callback_proto_v1_message_proto_goTypes,
DependencyIndexes: file_temporal_server_chasm_lib_callback_proto_v1_message_proto_depIdxs,
EnumInfos: file_temporal_server_chasm_lib_callback_proto_v1_message_proto_enumTypes,
MessageInfos: file_temporal_server_chasm_lib_callback_proto_v1_message_proto_msgTypes,
}.Build()
File_temporal_server_chasm_lib_callback_proto_v1_message_proto = out.File
file_temporal_server_chasm_lib_callback_proto_v1_message_proto_goTypes = nil
file_temporal_server_chasm_lib_callback_proto_v1_message_proto_depIdxs = nil
}
}
return file_temporal_server_api_enums_v1_nexus_proto_enumTypes[0].Descriptor()
}
func (NexusOperationState) Type() protoreflect.EnumType {
}
func file_temporal_server_api_enums_v1_nexus_proto_init() {
if File_temporal_server_api_enums_v1_nexus_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_enums_v1_nexus_proto_rawDesc), len(file_temporal_server_api_enums_v1_nexus_proto_rawDesc)),
NumEnums: 1,
NumMessages: 0,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_enums_v1_nexus_proto_goTypes,
DependencyIndexes: file_temporal_server_api_enums_v1_nexus_proto_depIdxs,
EnumInfos: file_temporal_server_api_enums_v1_nexus_proto_enumTypes,
}.Build()
File_temporal_server_api_enums_v1_nexus_proto = out.File
file_temporal_server_api_enums_v1_nexus_proto_goTypes = nil
file_temporal_server_api_enums_v1_nexus_proto_depIdxs = nil
}
}
return file_temporal_server_api_enums_v1_predicate_proto_enumTypes[0].Descriptor()
}
func (PredicateType) Type() protoreflect.EnumType {
}
func file_temporal_server_api_enums_v1_predicate_proto_init() {
if File_temporal_server_api_enums_v1_predicate_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_enums_v1_predicate_proto_rawDesc), len(file_temporal_server_api_enums_v1_predicate_proto_rawDesc)),
NumEnums: 1,
NumMessages: 0,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_enums_v1_predicate_proto_goTypes,
DependencyIndexes: file_temporal_server_api_enums_v1_predicate_proto_depIdxs,
EnumInfos: file_temporal_server_api_enums_v1_predicate_proto_enumTypes,
}.Build()
File_temporal_server_api_enums_v1_predicate_proto = out.File
file_temporal_server_api_enums_v1_predicate_proto_goTypes = nil
file_temporal_server_api_enums_v1_predicate_proto_depIdxs = nil
}
}
return file_temporal_server_api_enums_v1_workflow_task_type_proto_enumTypes[0].Descriptor()
}
func (WorkflowTaskType) Type() protoreflect.EnumType {
}
func init() { file_temporal_server_api_enums_v1_workflow_task_type_proto_init() }
workflow_task_type.pb.go
func file_temporal_server_api_enums_v1_workflow_task_type_proto_init() {
if File_temporal_server_api_enums_v1_workflow_task_type_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_enums_v1_workflow_task_type_proto_rawDesc), len(file_temporal_server_api_enums_v1_workflow_task_type_proto_rawDesc)),
NumEnums: 1,
NumMessages: 0,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_enums_v1_workflow_task_type_proto_goTypes,
DependencyIndexes: file_temporal_server_api_enums_v1_workflow_task_type_proto_depIdxs,
EnumInfos: file_temporal_server_api_enums_v1_workflow_task_type_proto_enumTypes,
}.Build()
File_temporal_server_api_enums_v1_workflow_task_type_proto = out.File
file_temporal_server_api_enums_v1_workflow_task_type_proto_goTypes = nil
file_temporal_server_api_enums_v1_workflow_task_type_proto_depIdxs = nil
}
)
if path := hasSharedStructure(reflect.ValueOf(def), "root"); path != "" {
sharedStructureWarnings.Store(key, path)
}
}
// If you see this warning, it means that a default value used in New*TypedSetting has a
// non-nil slice or map in it. That can lead to confusing behavior since the value from
// dynamic config will be merged over the default value (e.g. the slice will be appended
// to, not replaced). If that behavior is desired, you can avoid this warning by using
// New*TypedSettingWithConverter and referring to dynamicconfig.ConvertStructure
// explicitly. Otherwise use nil slices and maps, including at the top level
// (so `[]string(nil)` instead of `[]string{}`).
logSharedStructureWarningsOnce.Do(func() {
sharedStructureWarnings.Range(func(key, path any) bool {
softassert.Fail(logger,
"default value contains shared structure",
}
// nolint:exhaustive // deliberately not exhaustive
switch v.Kind() {
case reflect.Map, reflect.Slice, reflect.Pointer:
if !v.IsNil() {
return path
}
if !v.IsNil() {
return hasSharedStructure(v.Elem(), path)
}
for i := range v.NumField() {
if p := hasSharedStructure(v.Field(i), path+"."+v.Type().Field(i).Name); p != "" {
return p
}
}
func init() { file_temporal_server_api_common_v1_api_category_proto_init() }
api_category.pb.go
func file_temporal_server_api_common_v1_api_category_proto_init() {
if File_temporal_server_api_common_v1_api_category_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_common_v1_api_category_proto_rawDesc), len(file_temporal_server_api_common_v1_api_category_proto_rawDesc)),
NumEnums: 1,
NumMessages: 1,
NumExtensions: 1,
NumServices: 0,
},
GoTypes: file_temporal_server_api_common_v1_api_category_proto_goTypes,
DependencyIndexes: file_temporal_server_api_common_v1_api_category_proto_depIdxs,
EnumInfos: file_temporal_server_api_common_v1_api_category_proto_enumTypes,
MessageInfos: file_temporal_server_api_common_v1_api_category_proto_msgTypes,
ExtensionInfos: file_temporal_server_api_common_v1_api_category_proto_extTypes,
}.Build()
File_temporal_server_api_common_v1_api_category_proto = out.File
file_temporal_server_api_common_v1_api_category_proto_goTypes = nil
file_temporal_server_api_common_v1_api_category_proto_depIdxs = nil
}
}
func init() { file_temporal_server_chasm_lib_activity_proto_v1_request_response_proto_init() }
request_response.pb.go
func file_temporal_server_chasm_lib_activity_proto_v1_request_response_proto_init() {
if File_temporal_server_chasm_lib_activity_proto_v1_request_response_proto != nil {
return
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_chasm_lib_activity_proto_v1_request_response_proto_rawDesc), len(file_temporal_server_chasm_lib_activity_proto_v1_request_response_proto_rawDesc)),
NumEnums: 0,
NumMessages: 20,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_chasm_lib_activity_proto_v1_request_response_proto_goTypes,
DependencyIndexes: file_temporal_server_chasm_lib_activity_proto_v1_request_response_proto_depIdxs,
MessageInfos: file_temporal_server_chasm_lib_activity_proto_v1_request_response_proto_msgTypes,
}.Build()
File_temporal_server_chasm_lib_activity_proto_v1_request_response_proto = out.File
file_temporal_server_chasm_lib_activity_proto_v1_request_response_proto_goTypes = nil
file_temporal_server_chasm_lib_activity_proto_v1_request_response_proto_depIdxs = nil
}
}
func init() { file_temporal_server_chasm_lib_nexusoperation_proto_v1_request_response_proto_init() }
request_response.pb.go
func file_temporal_server_chasm_lib_nexusoperation_proto_v1_request_response_proto_init() {
if File_temporal_server_chasm_lib_nexusoperation_proto_v1_request_response_proto != nil {
return
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_chasm_lib_nexusoperation_proto_v1_request_response_proto_rawDesc), len(file_temporal_server_chasm_lib_nexusoperation_proto_v1_request_response_proto_rawDesc)),
NumEnums: 0,
NumMessages: 12,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_chasm_lib_nexusoperation_proto_v1_request_response_proto_goTypes,
DependencyIndexes: file_temporal_server_chasm_lib_nexusoperation_proto_v1_request_response_proto_depIdxs,
MessageInfos: file_temporal_server_chasm_lib_nexusoperation_proto_v1_request_response_proto_msgTypes,
}.Build()
File_temporal_server_chasm_lib_nexusoperation_proto_v1_request_response_proto = out.File
file_temporal_server_chasm_lib_nexusoperation_proto_v1_request_response_proto_goTypes = nil
file_temporal_server_chasm_lib_nexusoperation_proto_v1_request_response_proto_depIdxs = nil
}
// NewMockMetadata creates a new mock instance.
mock := &MockMetadata{ctrl: ctrl}
mock.recorder = &MockMetadataMockRecorder{mock}
return mock
}
// EXPECT returns an object that allows the caller to indicate expected use.
return m.recorder
}
// ClusterNameForFailoverVersion mocks base method.
// GetClusterID mocks base method.
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "GetClusterID")
ret0, _ := ret[0].(int64)
return ret0
}
// GetClusterID indicates an expected call of GetClusterID.
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetClusterID", reflect.TypeOf((*MockMetadata)(nil).GetClusterID))
}
// GetCurrentClusterName mocks base method.
// GetCurrentClusterName indicates an expected call of GetCurrentClusterName.
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetCurrentClusterName", reflect.TypeOf((*MockMetadata)(nil).GetCurrentClusterName))
}
// GetFailoverVersionIncrement mocks base method.
}
func init() { file_temporal_server_api_persistence_v1_task_queues_proto_init() }
task_queues.pb.go
func file_temporal_server_api_persistence_v1_task_queues_proto_init() {
if File_temporal_server_api_persistence_v1_task_queues_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_persistence_v1_task_queues_proto_rawDesc), len(file_temporal_server_api_persistence_v1_task_queues_proto_rawDesc)),
NumEnums: 1,
NumMessages: 13,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_persistence_v1_task_queues_proto_goTypes,
DependencyIndexes: file_temporal_server_api_persistence_v1_task_queues_proto_depIdxs,
EnumInfos: file_temporal_server_api_persistence_v1_task_queues_proto_enumTypes,
MessageInfos: file_temporal_server_api_persistence_v1_task_queues_proto_msgTypes,
}.Build()
File_temporal_server_api_persistence_v1_task_queues_proto = out.File
file_temporal_server_api_persistence_v1_task_queues_proto_goTypes = nil
file_temporal_server_api_persistence_v1_task_queues_proto_depIdxs = nil
}
}
func file_temporal_server_api_routing_v1_extension_proto_init() {
if File_temporal_server_api_routing_v1_extension_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_routing_v1_extension_proto_rawDesc), len(file_temporal_server_api_routing_v1_extension_proto_rawDesc)),
NumEnums: 0,
NumMessages: 1,
NumExtensions: 1,
NumServices: 0,
},
GoTypes: file_temporal_server_api_routing_v1_extension_proto_goTypes,
DependencyIndexes: file_temporal_server_api_routing_v1_extension_proto_depIdxs,
MessageInfos: file_temporal_server_api_routing_v1_extension_proto_msgTypes,
ExtensionInfos: file_temporal_server_api_routing_v1_extension_proto_extTypes,
}.Build()
File_temporal_server_api_routing_v1_extension_proto = out.File
file_temporal_server_api_routing_v1_extension_proto_goTypes = nil
file_temporal_server_api_routing_v1_extension_proto_depIdxs = nil
}
}
func init() { file_temporal_server_chasm_lib_activity_proto_v1_tasks_proto_init() }
tasks.pb.go
func file_temporal_server_chasm_lib_activity_proto_v1_tasks_proto_init() {
if File_temporal_server_chasm_lib_activity_proto_v1_tasks_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_chasm_lib_activity_proto_v1_tasks_proto_rawDesc), len(file_temporal_server_chasm_lib_activity_proto_v1_tasks_proto_rawDesc)),
NumEnums: 2,
NumMessages: 5,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_chasm_lib_activity_proto_v1_tasks_proto_goTypes,
DependencyIndexes: file_temporal_server_chasm_lib_activity_proto_v1_tasks_proto_depIdxs,
EnumInfos: file_temporal_server_chasm_lib_activity_proto_v1_tasks_proto_enumTypes,
MessageInfos: file_temporal_server_chasm_lib_activity_proto_v1_tasks_proto_msgTypes,
}.Build()
File_temporal_server_chasm_lib_activity_proto_v1_tasks_proto = out.File
file_temporal_server_chasm_lib_activity_proto_v1_tasks_proto_goTypes = nil
file_temporal_server_chasm_lib_activity_proto_v1_tasks_proto_depIdxs = nil
}
}
func file_temporal_server_api_adminservice_v1_service_proto_init() {
if File_temporal_server_api_adminservice_v1_service_proto != nil {
return
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_adminservice_v1_service_proto_rawDesc), len(file_temporal_server_api_adminservice_v1_service_proto_rawDesc)),
NumEnums: 0,
NumMessages: 0,
NumExtensions: 0,
NumServices: 1,
},
GoTypes: file_temporal_server_api_adminservice_v1_service_proto_goTypes,
DependencyIndexes: file_temporal_server_api_adminservice_v1_service_proto_depIdxs,
}.Build()
File_temporal_server_api_adminservice_v1_service_proto = out.File
file_temporal_server_api_adminservice_v1_service_proto_goTypes = nil
file_temporal_server_api_adminservice_v1_service_proto_depIdxs = nil
}
}
func file_temporal_server_api_archiver_v1_message_proto_init() {
if File_temporal_server_api_archiver_v1_message_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_archiver_v1_message_proto_rawDesc), len(file_temporal_server_api_archiver_v1_message_proto_rawDesc)),
NumEnums: 0,
NumMessages: 4,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_archiver_v1_message_proto_goTypes,
DependencyIndexes: file_temporal_server_api_archiver_v1_message_proto_depIdxs,
MessageInfos: file_temporal_server_api_archiver_v1_message_proto_msgTypes,
}.Build()
File_temporal_server_api_archiver_v1_message_proto = out.File
file_temporal_server_api_archiver_v1_message_proto_goTypes = nil
file_temporal_server_api_archiver_v1_message_proto_depIdxs = nil
}
}
func file_temporal_server_api_chasm_v1_message_proto_init() {
if File_temporal_server_api_chasm_v1_message_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_chasm_v1_message_proto_rawDesc), len(file_temporal_server_api_chasm_v1_message_proto_rawDesc)),
NumEnums: 0,
NumMessages: 1,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_chasm_v1_message_proto_goTypes,
DependencyIndexes: file_temporal_server_api_chasm_v1_message_proto_depIdxs,
MessageInfos: file_temporal_server_api_chasm_v1_message_proto_msgTypes,
}.Build()
File_temporal_server_api_chasm_v1_message_proto = out.File
file_temporal_server_api_chasm_v1_message_proto_goTypes = nil
file_temporal_server_api_chasm_v1_message_proto_depIdxs = nil
}
}
func file_temporal_server_api_checksum_v1_message_proto_init() {
if File_temporal_server_api_checksum_v1_message_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_checksum_v1_message_proto_rawDesc), len(file_temporal_server_api_checksum_v1_message_proto_rawDesc)),
NumEnums: 0,
NumMessages: 1,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_checksum_v1_message_proto_goTypes,
DependencyIndexes: file_temporal_server_api_checksum_v1_message_proto_depIdxs,
MessageInfos: file_temporal_server_api_checksum_v1_message_proto_msgTypes,
}.Build()
File_temporal_server_api_checksum_v1_message_proto = out.File
file_temporal_server_api_checksum_v1_message_proto_goTypes = nil
file_temporal_server_api_checksum_v1_message_proto_depIdxs = nil
}
}
func file_temporal_server_api_cluster_v1_message_proto_init() {
if File_temporal_server_api_cluster_v1_message_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_cluster_v1_message_proto_rawDesc), len(file_temporal_server_api_cluster_v1_message_proto_rawDesc)),
NumEnums: 0,
NumMessages: 4,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_cluster_v1_message_proto_goTypes,
DependencyIndexes: file_temporal_server_api_cluster_v1_message_proto_depIdxs,
MessageInfos: file_temporal_server_api_cluster_v1_message_proto_msgTypes,
}.Build()
File_temporal_server_api_cluster_v1_message_proto = out.File
file_temporal_server_api_cluster_v1_message_proto_goTypes = nil
file_temporal_server_api_cluster_v1_message_proto_depIdxs = nil
}
}
func file_temporal_server_api_common_v1_dlq_proto_init() {
if File_temporal_server_api_common_v1_dlq_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_common_v1_dlq_proto_rawDesc), len(file_temporal_server_api_common_v1_dlq_proto_rawDesc)),
NumEnums: 0,
NumMessages: 4,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_common_v1_dlq_proto_goTypes,
DependencyIndexes: file_temporal_server_api_common_v1_dlq_proto_depIdxs,
MessageInfos: file_temporal_server_api_common_v1_dlq_proto_msgTypes,
}.Build()
File_temporal_server_api_common_v1_dlq_proto = out.File
file_temporal_server_api_common_v1_dlq_proto_goTypes = nil
file_temporal_server_api_common_v1_dlq_proto_depIdxs = nil
}
}
func init() { file_temporal_server_api_contextpropagation_v1_message_proto_init() }
message.pb.go
func file_temporal_server_api_contextpropagation_v1_message_proto_init() {
if File_temporal_server_api_contextpropagation_v1_message_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_contextpropagation_v1_message_proto_rawDesc), len(file_temporal_server_api_contextpropagation_v1_message_proto_rawDesc)),
NumEnums: 0,
NumMessages: 2,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_contextpropagation_v1_message_proto_goTypes,
DependencyIndexes: file_temporal_server_api_contextpropagation_v1_message_proto_depIdxs,
MessageInfos: file_temporal_server_api_contextpropagation_v1_message_proto_msgTypes,
}.Build()
File_temporal_server_api_contextpropagation_v1_message_proto = out.File
file_temporal_server_api_contextpropagation_v1_message_proto_goTypes = nil
file_temporal_server_api_contextpropagation_v1_message_proto_depIdxs = nil
}
}
func file_temporal_server_api_deployment_v1_message_proto_init() {
if File_temporal_server_api_deployment_v1_message_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_deployment_v1_message_proto_rawDesc), len(file_temporal_server_api_deployment_v1_message_proto_rawDesc)),
NumEnums: 0,
NumMessages: 75,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_deployment_v1_message_proto_goTypes,
DependencyIndexes: file_temporal_server_api_deployment_v1_message_proto_depIdxs,
MessageInfos: file_temporal_server_api_deployment_v1_message_proto_msgTypes,
}.Build()
File_temporal_server_api_deployment_v1_message_proto = out.File
file_temporal_server_api_deployment_v1_message_proto_goTypes = nil
file_temporal_server_api_deployment_v1_message_proto_depIdxs = nil
}
}
func file_temporal_server_api_enums_v1_cluster_proto_init() {
if File_temporal_server_api_enums_v1_cluster_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_enums_v1_cluster_proto_rawDesc), len(file_temporal_server_api_enums_v1_cluster_proto_rawDesc)),
NumEnums: 2,
NumMessages: 0,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_enums_v1_cluster_proto_goTypes,
DependencyIndexes: file_temporal_server_api_enums_v1_cluster_proto_depIdxs,
EnumInfos: file_temporal_server_api_enums_v1_cluster_proto_enumTypes,
}.Build()
File_temporal_server_api_enums_v1_cluster_proto = out.File
file_temporal_server_api_enums_v1_cluster_proto_goTypes = nil
file_temporal_server_api_enums_v1_cluster_proto_depIdxs = nil
}
}
func file_temporal_server_api_enums_v1_dlq_proto_init() {
if File_temporal_server_api_enums_v1_dlq_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_enums_v1_dlq_proto_rawDesc), len(file_temporal_server_api_enums_v1_dlq_proto_rawDesc)),
NumEnums: 2,
NumMessages: 0,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_enums_v1_dlq_proto_goTypes,
DependencyIndexes: file_temporal_server_api_enums_v1_dlq_proto_depIdxs,
EnumInfos: file_temporal_server_api_enums_v1_dlq_proto_enumTypes,
}.Build()
File_temporal_server_api_enums_v1_dlq_proto = out.File
file_temporal_server_api_enums_v1_dlq_proto_goTypes = nil
file_temporal_server_api_enums_v1_dlq_proto_depIdxs = nil
}
}
func init() { file_temporal_server_api_enums_v1_fairness_state_proto_init() }
fairness_state.pb.go
func file_temporal_server_api_enums_v1_fairness_state_proto_init() {
if File_temporal_server_api_enums_v1_fairness_state_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_enums_v1_fairness_state_proto_rawDesc), len(file_temporal_server_api_enums_v1_fairness_state_proto_rawDesc)),
NumEnums: 1,
NumMessages: 0,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_enums_v1_fairness_state_proto_goTypes,
DependencyIndexes: file_temporal_server_api_enums_v1_fairness_state_proto_depIdxs,
EnumInfos: file_temporal_server_api_enums_v1_fairness_state_proto_enumTypes,
}.Build()
File_temporal_server_api_enums_v1_fairness_state_proto = out.File
file_temporal_server_api_enums_v1_fairness_state_proto_goTypes = nil
file_temporal_server_api_enums_v1_fairness_state_proto_depIdxs = nil
}
}
func file_temporal_server_api_enums_v1_replication_proto_init() {
if File_temporal_server_api_enums_v1_replication_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_enums_v1_replication_proto_rawDesc), len(file_temporal_server_api_enums_v1_replication_proto_rawDesc)),
NumEnums: 3,
NumMessages: 0,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_enums_v1_replication_proto_goTypes,
DependencyIndexes: file_temporal_server_api_enums_v1_replication_proto_depIdxs,
EnumInfos: file_temporal_server_api_enums_v1_replication_proto_enumTypes,
}.Build()
File_temporal_server_api_enums_v1_replication_proto = out.File
file_temporal_server_api_enums_v1_replication_proto_goTypes = nil
file_temporal_server_api_enums_v1_replication_proto_depIdxs = nil
}
}
func file_temporal_server_api_errordetails_v1_message_proto_init() {
if File_temporal_server_api_errordetails_v1_message_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_errordetails_v1_message_proto_rawDesc), len(file_temporal_server_api_errordetails_v1_message_proto_rawDesc)),
NumEnums: 0,
NumMessages: 10,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_errordetails_v1_message_proto_goTypes,
DependencyIndexes: file_temporal_server_api_errordetails_v1_message_proto_depIdxs,
MessageInfos: file_temporal_server_api_errordetails_v1_message_proto_msgTypes,
}.Build()
File_temporal_server_api_errordetails_v1_message_proto = out.File
file_temporal_server_api_errordetails_v1_message_proto_goTypes = nil
file_temporal_server_api_errordetails_v1_message_proto_depIdxs = nil
}
}
func file_temporal_server_api_health_v1_message_proto_init() {
if File_temporal_server_api_health_v1_message_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_health_v1_message_proto_rawDesc), len(file_temporal_server_api_health_v1_message_proto_rawDesc)),
NumEnums: 0,
NumMessages: 3,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_health_v1_message_proto_goTypes,
DependencyIndexes: file_temporal_server_api_health_v1_message_proto_depIdxs,
MessageInfos: file_temporal_server_api_health_v1_message_proto_msgTypes,
}.Build()
File_temporal_server_api_health_v1_message_proto = out.File
file_temporal_server_api_health_v1_message_proto_goTypes = nil
file_temporal_server_api_health_v1_message_proto_depIdxs = nil
}
}
func file_temporal_server_api_historyservice_v1_service_proto_init() {
if File_temporal_server_api_historyservice_v1_service_proto != nil {
return
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_historyservice_v1_service_proto_rawDesc), len(file_temporal_server_api_historyservice_v1_service_proto_rawDesc)),
NumEnums: 0,
NumMessages: 0,
NumExtensions: 0,
NumServices: 1,
},
GoTypes: file_temporal_server_api_historyservice_v1_service_proto_goTypes,
DependencyIndexes: file_temporal_server_api_historyservice_v1_service_proto_depIdxs,
}.Build()
File_temporal_server_api_historyservice_v1_service_proto = out.File
file_temporal_server_api_historyservice_v1_service_proto_goTypes = nil
file_temporal_server_api_historyservice_v1_service_proto_depIdxs = nil
}
}
func init() { file_temporal_server_api_matchingservice_v1_service_proto_init() }
service.pb.go
func file_temporal_server_api_matchingservice_v1_service_proto_init() {
if File_temporal_server_api_matchingservice_v1_service_proto != nil {
return
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_matchingservice_v1_service_proto_rawDesc), len(file_temporal_server_api_matchingservice_v1_service_proto_rawDesc)),
NumEnums: 0,
NumMessages: 0,
NumExtensions: 0,
NumServices: 1,
},
GoTypes: file_temporal_server_api_matchingservice_v1_service_proto_goTypes,
DependencyIndexes: file_temporal_server_api_matchingservice_v1_service_proto_depIdxs,
}.Build()
File_temporal_server_api_matchingservice_v1_service_proto = out.File
file_temporal_server_api_matchingservice_v1_service_proto_goTypes = nil
file_temporal_server_api_matchingservice_v1_service_proto_depIdxs = nil
}
}
func file_temporal_server_api_metrics_v1_message_proto_init() {
if File_temporal_server_api_metrics_v1_message_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_metrics_v1_message_proto_rawDesc), len(file_temporal_server_api_metrics_v1_message_proto_rawDesc)),
NumEnums: 0,
NumMessages: 2,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_metrics_v1_message_proto_goTypes,
DependencyIndexes: file_temporal_server_api_metrics_v1_message_proto_depIdxs,
MessageInfos: file_temporal_server_api_metrics_v1_message_proto_msgTypes,
}.Build()
File_temporal_server_api_metrics_v1_message_proto = out.File
file_temporal_server_api_metrics_v1_message_proto_goTypes = nil
file_temporal_server_api_metrics_v1_message_proto_depIdxs = nil
}
}
func file_temporal_server_api_namespace_v1_message_proto_init() {
if File_temporal_server_api_namespace_v1_message_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_namespace_v1_message_proto_rawDesc), len(file_temporal_server_api_namespace_v1_message_proto_rawDesc)),
NumEnums: 0,
NumMessages: 1,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_namespace_v1_message_proto_goTypes,
DependencyIndexes: file_temporal_server_api_namespace_v1_message_proto_depIdxs,
MessageInfos: file_temporal_server_api_namespace_v1_message_proto_msgTypes,
}.Build()
File_temporal_server_api_namespace_v1_message_proto = out.File
file_temporal_server_api_namespace_v1_message_proto_goTypes = nil
file_temporal_server_api_namespace_v1_message_proto_depIdxs = nil
}
}
func init() { file_temporal_server_api_persistence_v1_chasm_visibility_proto_init() }
chasm_visibility.pb.go
func file_temporal_server_api_persistence_v1_chasm_visibility_proto_init() {
if File_temporal_server_api_persistence_v1_chasm_visibility_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_persistence_v1_chasm_visibility_proto_rawDesc), len(file_temporal_server_api_persistence_v1_chasm_visibility_proto_rawDesc)),
NumEnums: 0,
NumMessages: 2,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_persistence_v1_chasm_visibility_proto_goTypes,
DependencyIndexes: file_temporal_server_api_persistence_v1_chasm_visibility_proto_depIdxs,
MessageInfos: file_temporal_server_api_persistence_v1_chasm_visibility_proto_msgTypes,
}.Build()
File_temporal_server_api_persistence_v1_chasm_visibility_proto = out.File
file_temporal_server_api_persistence_v1_chasm_visibility_proto_goTypes = nil
file_temporal_server_api_persistence_v1_chasm_visibility_proto_depIdxs = nil
}
}
func init() { file_temporal_server_api_persistence_v1_cluster_metadata_proto_init() }
cluster_metadata.pb.go
func file_temporal_server_api_persistence_v1_cluster_metadata_proto_init() {
if File_temporal_server_api_persistence_v1_cluster_metadata_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_persistence_v1_cluster_metadata_proto_rawDesc), len(file_temporal_server_api_persistence_v1_cluster_metadata_proto_rawDesc)),
NumEnums: 0,
NumMessages: 5,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_persistence_v1_cluster_metadata_proto_goTypes,
DependencyIndexes: file_temporal_server_api_persistence_v1_cluster_metadata_proto_depIdxs,
MessageInfos: file_temporal_server_api_persistence_v1_cluster_metadata_proto_msgTypes,
}.Build()
File_temporal_server_api_persistence_v1_cluster_metadata_proto = out.File
file_temporal_server_api_persistence_v1_cluster_metadata_proto_goTypes = nil
file_temporal_server_api_persistence_v1_cluster_metadata_proto_depIdxs = nil
}
}
func init() { file_temporal_server_api_persistence_v1_namespaces_proto_init() }
namespaces.pb.go
func file_temporal_server_api_persistence_v1_namespaces_proto_init() {
if File_temporal_server_api_persistence_v1_namespaces_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_persistence_v1_namespaces_proto_rawDesc), len(file_temporal_server_api_persistence_v1_namespaces_proto_rawDesc)),
NumEnums: 0,
NumMessages: 8,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_persistence_v1_namespaces_proto_goTypes,
DependencyIndexes: file_temporal_server_api_persistence_v1_namespaces_proto_depIdxs,
MessageInfos: file_temporal_server_api_persistence_v1_namespaces_proto_msgTypes,
}.Build()
File_temporal_server_api_persistence_v1_namespaces_proto = out.File
file_temporal_server_api_persistence_v1_namespaces_proto_goTypes = nil
file_temporal_server_api_persistence_v1_namespaces_proto_depIdxs = nil
}
}
func init() { file_temporal_server_api_persistence_v1_queue_metadata_proto_init() }
queue_metadata.pb.go
func file_temporal_server_api_persistence_v1_queue_metadata_proto_init() {
if File_temporal_server_api_persistence_v1_queue_metadata_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_persistence_v1_queue_metadata_proto_rawDesc), len(file_temporal_server_api_persistence_v1_queue_metadata_proto_rawDesc)),
NumEnums: 0,
NumMessages: 2,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_persistence_v1_queue_metadata_proto_goTypes,
DependencyIndexes: file_temporal_server_api_persistence_v1_queue_metadata_proto_depIdxs,
MessageInfos: file_temporal_server_api_persistence_v1_queue_metadata_proto_msgTypes,
}.Build()
File_temporal_server_api_persistence_v1_queue_metadata_proto = out.File
file_temporal_server_api_persistence_v1_queue_metadata_proto_goTypes = nil
file_temporal_server_api_persistence_v1_queue_metadata_proto_depIdxs = nil
}
}
func file_temporal_server_api_persistence_v1_tasks_proto_init() {
if File_temporal_server_api_persistence_v1_tasks_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_persistence_v1_tasks_proto_rawDesc), len(file_temporal_server_api_persistence_v1_tasks_proto_rawDesc)),
NumEnums: 0,
NumMessages: 8,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_persistence_v1_tasks_proto_goTypes,
DependencyIndexes: file_temporal_server_api_persistence_v1_tasks_proto_depIdxs,
MessageInfos: file_temporal_server_api_persistence_v1_tasks_proto_msgTypes,
}.Build()
File_temporal_server_api_persistence_v1_tasks_proto = out.File
file_temporal_server_api_persistence_v1_tasks_proto_goTypes = nil
file_temporal_server_api_persistence_v1_tasks_proto_depIdxs = nil
}
}
func file_temporal_server_api_token_v1_message_proto_init() {
if File_temporal_server_api_token_v1_message_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_token_v1_message_proto_rawDesc), len(file_temporal_server_api_token_v1_message_proto_rawDesc)),
NumEnums: 0,
NumMessages: 7,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_token_v1_message_proto_goTypes,
DependencyIndexes: file_temporal_server_api_token_v1_message_proto_depIdxs,
MessageInfos: file_temporal_server_api_token_v1_message_proto_msgTypes,
}.Build()
File_temporal_server_api_token_v1_message_proto = out.File
file_temporal_server_api_token_v1_message_proto_goTypes = nil
file_temporal_server_api_token_v1_message_proto_depIdxs = nil
}
}
func init() { file_temporal_server_api_visibilityservice_v1_request_response_proto_init() }
request_response.pb.go
func file_temporal_server_api_visibilityservice_v1_request_response_proto_init() {
if File_temporal_server_api_visibilityservice_v1_request_response_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_visibilityservice_v1_request_response_proto_rawDesc), len(file_temporal_server_api_visibilityservice_v1_request_response_proto_rawDesc)),
NumEnums: 0,
NumMessages: 5,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_visibilityservice_v1_request_response_proto_goTypes,
DependencyIndexes: file_temporal_server_api_visibilityservice_v1_request_response_proto_depIdxs,
MessageInfos: file_temporal_server_api_visibilityservice_v1_request_response_proto_msgTypes,
}.Build()
File_temporal_server_api_visibilityservice_v1_request_response_proto = out.File
file_temporal_server_api_visibilityservice_v1_request_response_proto_goTypes = nil
file_temporal_server_api_visibilityservice_v1_request_response_proto_depIdxs = nil
}
}
func init() { file_temporal_server_chasm_lib_activity_proto_v1_service_proto_init() }
service.pb.go
func file_temporal_server_chasm_lib_activity_proto_v1_service_proto_init() {
if File_temporal_server_chasm_lib_activity_proto_v1_service_proto != nil {
return
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_chasm_lib_activity_proto_v1_service_proto_rawDesc), len(file_temporal_server_chasm_lib_activity_proto_v1_service_proto_rawDesc)),
NumEnums: 0,
NumMessages: 0,
NumExtensions: 0,
NumServices: 1,
},
GoTypes: file_temporal_server_chasm_lib_activity_proto_v1_service_proto_goTypes,
DependencyIndexes: file_temporal_server_chasm_lib_activity_proto_v1_service_proto_depIdxs,
}.Build()
File_temporal_server_chasm_lib_activity_proto_v1_service_proto = out.File
file_temporal_server_chasm_lib_activity_proto_v1_service_proto_goTypes = nil
file_temporal_server_chasm_lib_activity_proto_v1_service_proto_depIdxs = nil
}
}
func init() { file_temporal_server_chasm_lib_callback_proto_v1_tasks_proto_init() }
tasks.pb.go
func file_temporal_server_chasm_lib_callback_proto_v1_tasks_proto_init() {
if File_temporal_server_chasm_lib_callback_proto_v1_tasks_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_chasm_lib_callback_proto_v1_tasks_proto_rawDesc), len(file_temporal_server_chasm_lib_callback_proto_v1_tasks_proto_rawDesc)),
NumEnums: 0,
NumMessages: 2,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_chasm_lib_callback_proto_v1_tasks_proto_goTypes,
DependencyIndexes: file_temporal_server_chasm_lib_callback_proto_v1_tasks_proto_depIdxs,
MessageInfos: file_temporal_server_chasm_lib_callback_proto_v1_tasks_proto_msgTypes,
}.Build()
File_temporal_server_chasm_lib_callback_proto_v1_tasks_proto = out.File
file_temporal_server_chasm_lib_callback_proto_v1_tasks_proto_goTypes = nil
file_temporal_server_chasm_lib_callback_proto_v1_tasks_proto_depIdxs = nil
}
}
func init() { file_temporal_server_chasm_lib_nexusoperation_proto_v1_service_proto_init() }
service.pb.go
func file_temporal_server_chasm_lib_nexusoperation_proto_v1_service_proto_init() {
if File_temporal_server_chasm_lib_nexusoperation_proto_v1_service_proto != nil {
return
}
file_temporal_server_chasm_lib_nexusoperation_proto_v1_request_response_proto_init()
service.pb.go
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_chasm_lib_nexusoperation_proto_v1_service_proto_rawDesc), len(file_temporal_server_chasm_lib_nexusoperation_proto_v1_service_proto_rawDesc)),
NumEnums: 0,
NumMessages: 0,
NumExtensions: 0,
NumServices: 1,
},
GoTypes: file_temporal_server_chasm_lib_nexusoperation_proto_v1_service_proto_goTypes,
DependencyIndexes: file_temporal_server_chasm_lib_nexusoperation_proto_v1_service_proto_depIdxs,
}.Build()
File_temporal_server_chasm_lib_nexusoperation_proto_v1_service_proto = out.File
file_temporal_server_chasm_lib_nexusoperation_proto_v1_service_proto_goTypes = nil
file_temporal_server_chasm_lib_nexusoperation_proto_v1_service_proto_depIdxs = nil
}
}
func init() { file_temporal_server_chasm_lib_nexusoperation_proto_v1_tasks_proto_init() }
tasks.pb.go
func file_temporal_server_chasm_lib_nexusoperation_proto_v1_tasks_proto_init() {
if File_temporal_server_chasm_lib_nexusoperation_proto_v1_tasks_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_chasm_lib_nexusoperation_proto_v1_tasks_proto_rawDesc), len(file_temporal_server_chasm_lib_nexusoperation_proto_v1_tasks_proto_rawDesc)),
NumEnums: 0,
NumMessages: 7,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_chasm_lib_nexusoperation_proto_v1_tasks_proto_goTypes,
DependencyIndexes: file_temporal_server_chasm_lib_nexusoperation_proto_v1_tasks_proto_depIdxs,
MessageInfos: file_temporal_server_chasm_lib_nexusoperation_proto_v1_tasks_proto_msgTypes,
}.Build()
File_temporal_server_chasm_lib_nexusoperation_proto_v1_tasks_proto = out.File
file_temporal_server_chasm_lib_nexusoperation_proto_v1_tasks_proto_goTypes = nil
file_temporal_server_chasm_lib_nexusoperation_proto_v1_tasks_proto_depIdxs = nil
}
}
func init() { file_temporal_server_chasm_lib_workflow_proto_v1_state_proto_init() }
state.pb.go
func file_temporal_server_chasm_lib_workflow_proto_v1_state_proto_init() {
if File_temporal_server_chasm_lib_workflow_proto_v1_state_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_chasm_lib_workflow_proto_v1_state_proto_rawDesc), len(file_temporal_server_chasm_lib_workflow_proto_v1_state_proto_rawDesc)),
NumEnums: 0,
NumMessages: 3,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_chasm_lib_workflow_proto_v1_state_proto_goTypes,
DependencyIndexes: file_temporal_server_chasm_lib_workflow_proto_v1_state_proto_depIdxs,
MessageInfos: file_temporal_server_chasm_lib_workflow_proto_v1_state_proto_msgTypes,
}.Build()
File_temporal_server_chasm_lib_workflow_proto_v1_state_proto = out.File
file_temporal_server_chasm_lib_workflow_proto_v1_state_proto_goTypes = nil
file_temporal_server_chasm_lib_workflow_proto_v1_state_proto_depIdxs = nil
}
}
func init() { file_temporal_server_chasm_lib_workflow_proto_v1_update_state_proto_init() }
update_state.pb.go
func file_temporal_server_chasm_lib_workflow_proto_v1_update_state_proto_init() {
if File_temporal_server_chasm_lib_workflow_proto_v1_update_state_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_chasm_lib_workflow_proto_v1_update_state_proto_rawDesc), len(file_temporal_server_chasm_lib_workflow_proto_v1_update_state_proto_rawDesc)),
NumEnums: 0,
NumMessages: 1,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_chasm_lib_workflow_proto_v1_update_state_proto_goTypes,
DependencyIndexes: file_temporal_server_chasm_lib_workflow_proto_v1_update_state_proto_depIdxs,
MessageInfos: file_temporal_server_chasm_lib_workflow_proto_v1_update_state_proto_msgTypes,
}.Build()
File_temporal_server_chasm_lib_workflow_proto_v1_update_state_proto = out.File
file_temporal_server_chasm_lib_workflow_proto_v1_update_state_proto_goTypes = nil
file_temporal_server_chasm_lib_workflow_proto_v1_update_state_proto_depIdxs = nil
}
}
hex.Encode(dst, u[:4])
dst[8] = '-'
hex.Encode(dst[9:13], u[4:6])
dst[13] = '-'
hex.Encode(dst[14:18], u[6:8])
dst[18] = '-'
hex.Encode(dst[19:23], u[8:10])
dst[23] = '-'
hex.Encode(dst[24:], u[10:])
}
// WildCardStringToRegexps converts a given slices of string patterns to a slice of regular expressions matching
// wildcards (*) with any substring.
var result strings.Builder
result.WriteRune('^')
for i, pattern := range patterns {
result.WriteRune('(')
first := true
for literal := range strings.SplitSeq(pattern, "*") {
if !first {
// Replace * with .*
result.WriteString(".*")
}
first = false
}
if i < len(patterns)-1 {
}
}
return regexp.Compile(result.String())
}
// MustWildCardStringsToRegexp is like WildCardStringsToRegexp but panics on error.
re, err := WildCardStringsToRegexp(patterns)
if err != nil {
panic(err) //nolint:forbidigo // Must* functions conventionally panic on error.
}
}
)
func newTaskRequestTracker(registry tasks.TaskCategoryRegistry) *taskRequestTracker {
task_request_tracker.go
outstandingTaskKeys := make(map[tasks.Category]map[tasks.Key]struct{})
for _, category := range registry.GetCategories() {
outstandingTaskKeys[category] = make(map[tasks.Key]struct{})
}
return &taskRequestTracker{
pendingTaskKeys: outstandingTaskKeys,
}
}
}
t.Lock()
defer t.Unlock()
for category := range t.pendingTaskKeys {
t.pendingTaskKeys[category] = make(map[tasks.Key]struct{})
}
t.inflightRequestCount = 0
t.closeWaitChannelsLocked()
}
for _, waitCh := range t.waitChannels {
close(waitCh)
}
}
)
var value T
return &FutureImpl[T]{
status: pending,
readyCh: make(chan struct{}),
value: value,
err: nil,
}
}
func (f *FutureImpl[T]) Get(
ctx context.Context,
if f.Ready() {
return f.value, f.err
}
case <-f.readyCh:
return f.value, f.err
var value T
return value, ctx.Err()
}
}
}
return atomic.LoadInt32(&f.status) == ready
}
// NewMockRegistry creates a new mock instance.
mock := &MockRegistry{ctrl: ctrl}
mock.recorder = &MockRegistryMockRecorder{mock}
return mock
}
// EXPECT returns an object that allows the caller to indicate expected use.
return m.recorder
}
// GetAllNamespaces mocks base method.
// GetNamespaceByID mocks base method.
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "GetNamespaceByID", id)
ret0, _ := ret[0].(*Namespace)
ret1, _ := ret[1].(error)
return ret0, ret1
}
// GetNamespaceByID indicates an expected call of GetNamespaceByID.
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetNamespaceByID", reflect.TypeOf((*MockRegistry)(nil).GetNamespaceByID), id)
}
// GetNamespaceByIDWithOptions mocks base method.
// NewRegistry creates a new [Registry].
return &Registry{
machines: make(map[string]StateMachineDefinition),
tasks: make(map[string]TaskSerializer),
immediateExecutors: make(map[string]any),
timerExecutors: make(map[string]any),
remoteExecutors: make(map[string]remoteMethodDefinition),
events: make(map[enumspb.EventType]EventDefinition),
}
}
// RegisterMachine registers a [StateMachineDefinition] by its type.
// Returns an [ErrDuplicateRegistration] if the state machine type has already been registered.
t := sm.Type()
if existing, ok := r.machines[t]; ok {
return fmt.Errorf("%w: state machine already registered for %v - %v", ErrDuplicateRegistration, sm.Type(), existing.Type())
}
return nil
}
// Machine returns a [StateMachineDefinition] for a given type and a boolean indicating whether it was found.
def, ok = r.machines[t]
return
}
// RegisterTaskSerializer registers a [TaskSerializer] for a given type.
)
var defaultProvider TaskGeneratorProvider = new(taskGeneratorProviderImpl)
populateTaskGeneratorProvider(defaultProvider)
}
func populateTaskGeneratorProvider(provider TaskGeneratorProvider) {
task_generator_provider.go
_taskGeneratorProvider.Store(&provider)
}
return *_taskGeneratorProvider.Load()
}
func (p *taskGeneratorProviderImpl) NewTaskGenerator(
shard historyi.ShardContext,
mutableState historyi.MutableState,
return NewTaskGenerator(
shard.GetNamespaceRegistry(),
mutableState,
shard.GetConfig(),
shard.GetArchivalMetadata(),
shard.GetLogger(),
)
}
}
return t.field
}
func (t ZapTag) Key() string {
}
return ZapTag{
field: zap.String(key, value),
}
}
func NewStringsTag(key string, value []string) ZapTag {
}
return ZapTag{
field: zap.Int64(key, value),
}
}
func NewInt(key string, value int) ZapTag {
}
return ZapTag{
field: zap.Bool(key, value),
}
}
func NewErrorTag(key string, value error) ZapTag {
}
func NewDefaultReplicationResolverFactory() ReplicationResolverFactory {
replication_resolver.go
return func(detail *persistencespb.NamespaceDetail) ReplicationResolver {
// By convention, a namespace with non-zero failover version is a global namespace
// This can be overridden by WithGlobalFlag mutation if needed
isGlobal := detail.FailoverVersion != 0
return &defaultReplicationResolver{
replicationConfig: detail.ReplicationConfig,
isGlobalNamespace: isGlobal,
failoverVersion: detail.FailoverVersion,
failoverNotificationVersion: detail.FailoverNotificationVersion,
}
}
}
}
func (r *defaultReplicationResolver) FailoverVersion(businessID string) int64 {
replication_resolver.go
return r.failoverVersion
}
func (r *defaultReplicationResolver) FailoverNotificationVersion() int64 {
}
r.isGlobalNamespace = isGlobal
}
func (r *defaultReplicationResolver) SetActiveCluster(clusterName string) {
)
func NewHistoryBranchUtil(serializer serialization.Serializer) *HistoryBranchUtilImpl {
history_branch_util.go
return &HistoryBranchUtilImpl{
serializer: serializer,
}
}
func (u *HistoryBranchUtilImpl) NewHistoryBranch(
_ time.Duration, // executionTimeout
_ time.Duration, // retentionDuration
var id string
if branchID == nil {
id = *branchID
}
TreeId: treeID,
BranchId: id,
Ancestors: ancestors,
}
data, err := u.serializer.HistoryBranchToBlob(bi)
if err != nil {
return nil, err
}
}
// each entry point that uses it. Essentially, get it from the dependency graph instead of calling this method, unless
// you're in a test.
return &MutableTaskCategoryRegistry{
categories: map[int]Category{
CategoryTransfer.ID(): CategoryTransfer,
CategoryTimer.ID(): CategoryTimer,
CategoryVisibility.ID(): CategoryVisibility,
CategoryReplication.ID(): CategoryReplication,
CategoryMemoryTimer.ID(): CategoryMemoryTimer,
CategoryOutbound.ID(): CategoryOutbound,
},
}
}
// AddCategory register a Category with the registry or panics if a Category with the same ID has already been
// registered.
if category, ok := r.categories[c.id]; ok {
panic(fmt.Sprintf(
"category id: %v has already been defined as type %v and name %v",
// GetCategories returns a deep copy of all registered Category objects from the registry.
func (r *MutableTaskCategoryRegistry) GetCategories() map[int]Category {
task_category_registry.go
return maps.Clone(r.categories)
}
// CalculateExternalPayloadSize calculates the total size and count of all external payloads in the given history events.
func CalculateExternalPayloadSize(events []*historypb.HistoryEvent, metricsHandler metrics.Handler) (size int64, count int64, err error) {
external_payload_size.go
var totalSize int64
var totalCount int64
visitor := func(vpc *proxy.VisitPayloadsContext, payloads []*commonpb.Payload) ([]*commonpb.Payload, error) {
for _, extPayload := range p.ExternalPayloads {
metricsHandler.Histogram(metrics.ExternalPayloadUploadSize.Name(), metrics.Bytes).Record(int64(extPayload.SizeBytes))
}
}
}
err := proxy.VisitPayloads(context.Background(), event, proxy.VisitPayloadsOptions{
external_payload_size.go
Visitor: visitor,
SkipSearchAttributes: true,
})
if err != nil {
return 0, 0, err
}
}
}
// NewMockTaskRefresher creates a new mock instance.
mock := &MockTaskRefresher{ctrl: ctrl}
mock.recorder = &MockTaskRefresherMockRecorder{mock}
return mock
}
// EXPECT returns an object that allows the caller to indicate expected use.
return m.recorder
}
// PartialRefresh mocks base method.
// Refresh mocks base method.
func (m *MockTaskRefresher) Refresh(ctx context.Context, mutableState interfaces.MutableState, shouldSkipGeneratingCloseTransferTask bool) error {
task_refresher_mock.go
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "Refresh", ctx, mutableState, shouldSkipGeneratingCloseTransferTask)
ret0, _ := ret[0].(error)
return ret0
}
// Refresh indicates an expected call of Refresh.
func (mr *MockTaskRefresherMockRecorder) Refresh(ctx, mutableState, shouldSkipGeneratingCloseTransferTask any) *gomock.Call {
task_refresher_mock.go
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Refresh", reflect.TypeOf((*MockTaskRefresher)(nil).Refresh), ctx, mutableState, shouldSkipGeneratingCloseTransferTask)
}
// encodingTypeFromEnv returns an EncodingType based on the environment variable `TEMPORAL_TEST_DATA_ENCODING`.
// It defaults to "ENCODING_TYPE_PROTO3" codec if the environment variable is not set.
codecType := os.Getenv(SerializerDataEncodingEnvVar)
switch strings.ToLower(codecType) {
return enumspb.ENCODING_TYPE_PROTO3
case "json":
return enumspb.ENCODING_TYPE_JSON
}
case enumspb.ENCODING_TYPE_JSON:
blob, err := codec.NewJSONPBEncoder().Encode(m)
EncodingType: enumspb.ENCODING_TYPE_JSON,
}, nil
data, err := proto.MarshalOptions{Deterministic: opts.deterministic}.Marshal(m)
if err != nil {
return nil, NewSerializationError(enumspb.ENCODING_TYPE_PROTO3, err)
}
EncodingType: enumspb.ENCODING_TYPE_PROTO3,
Data: data,
}, nil
default:
return nil, NewUnknownEncodingTypeError(encoding.String(), enumspb.ENCODING_TYPE_JSON, enumspb.ENCODING_TYPE_PROTO3)
)
if d == nil {
}
return d.AsDuration()
}
return durationpb.New(td)
}
func MinDurationPtr(d1 *durationpb.Duration, d2 *durationpb.Duration) *durationpb.Duration {
}
return durationMultipleOf(s, time.Second)
}
func DurationFromMinutes(m int64) *durationpb.Duration {
}
return durationMultipleOf(int64(d), time.Hour*24)
}
return DurationPtr(time.Duration(amt) * mult)
}
// ValidateAndCapProtoDuration validates protobuf durations for two conditions:
}
return ProtoAssertions{t}
}
// ProtoEqual compares two proto messages for equality using proto semantics. Options can be passed to customize
// comparison behavior, e.g. protorequire.IgnoreFields to exclude specific fields.
func ProtoEqual(t require.TestingT, a proto.Message, b proto.Message, opts ...Option) {
require.go
if th, ok := t.(helper); ok {
}
for _, opt := range opts {
opt(a, cfg)
}
if diff := cmp.Diff(a, b, cmpOpts...); diff != "" {
require.Fail(t, fmt.Sprintf("Proto mismatch (-want +got):\n%v", diff))
}
}
func (x ProtoAssertions) ProtoEqual(a proto.Message, b proto.Message, opts ...Option) {
require.go
if th, ok := x.t.(helper); ok {
}
}
// NewMockCache creates a new mock instance.
mock := &MockCache{ctrl: ctrl}
mock.recorder = &MockCacheMockRecorder{mock}
return mock
}
// EXPECT returns an object that allows the caller to indicate expected use.
return m.recorder
}
// DeleteEvent mocks base method.
// PutEvent mocks base method.
m.ctrl.T.Helper()
m.ctrl.Call(m, "PutEvent", key, event)
}
// PutEvent indicates an expected call of PutEvent.
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "PutEvent", reflect.TypeOf((*MockCache)(nil).PutEvent), key, event)
}
}
return &Registry{
libraries: make(map[string]Library),
rcByFqn: make(map[string]*RegistrableComponent),
rcByID: make(map[uint32]*RegistrableComponent),
rcByGoType: make(map[reflect.Type]*RegistrableComponent),
rtByFqn: make(map[string]*RegistrableTask),
rtByID: make(map[uint32]*RegistrableTask),
rtByGoType: make(map[reflect.Type]*RegistrableTask),
rcContextValues: make(map[any]valueWithFqn),
nexusServices: make(map[string]*nexus.Service),
NexusEndpointProcessor: NewNexusEndpointProcessor(),
logger: logger,
}
}
func (r *Registry) Register(lib Library) error {
// ContextMetadataGetMarkedActivityIDs returns the marked activity IDs from the context.
metadataCtx := getMetadataContext(ctx)
if metadataCtx == nil {
}
metadataCtx.Lock()
// getMetadataContext extracts metadata context from golang context.
metadataCtx := ctx.Value(metadataCtxKey)
if metadataCtx == nil {
}
mc, ok := metadataCtx.(*metadataContext)
if !ok {
// ContextMetadataSet sets a metadata key-value pair in the context, overwriting any existing value.
metadataCtx := getMetadataContext(ctx)
if metadataCtx == nil {
}
metadataCtx.Lock()
}
return &colName{Name: name}
}
func newSAColName(
fieldName string,
valueType enumspb.IndexedValueType,
return &saColName{
dbColName: newColName(dbColName),
alias: alias,
fieldName: fieldName,
valueType: valueType,
}
}
func newFuncExpr(name string, exprs ...sqlparser.Expr) *sqlparser.FuncExpr {
}
t, _ := time.Parse(time.RFC3339, "9999-12-31T23:59:59Z")
return t
}
// formatComparisonExprStringForError formats comparison expression after
)
buildInfo, ok := debug.ReadBuildInfo()
if !ok {
return
}
InfoData.GoVersion = buildInfo.GoVersion
for _, setting := range buildInfo.Settings {
switch setting.Key {
case "GOARCH":
InfoData.GoArch = setting.Value
case "GOOS":
InfoData.GoOs = setting.Value
case "CGO_ENABLED":
InfoData.CgoEnabled = setting.Value == "1"
case "vcs.revision":
InfoData.GitRevision = setting.Value
// StaticGradualChange returns a GradualChange whose Value always returns def and whose When
// always returns a time in the past.
return GradualChange[T]{New: def}
}
// Value returns the value for the given key at the given time.
// of type GradualChange into a GradualChange.
// nolint:revive // cognitive-complexity // this looks complicated but each case is fairly simple
func ConvertGradualChange[T any](def T) func(v any) (GradualChange[T], error) {
gradual_change.go
changeConverter := ConvertStructure(StaticGradualChange(def))
// Call this once so that if it's going to panic, it panics at static init time.
_, _ = changeConverter(nil)
switch reflect.TypeFor[T]() {
case reflect.TypeFor[bool]():
return func(v any) (GradualChange[T], error) {
if b, err := convertBool(v); err == nil {
var change GradualChange[T]
return changeConverter(v)
}
return func(v any) (GradualChange[T], error) {
if i, err := convertInt(v); err == nil {
var change GradualChange[T]
workflowID string,
runID string,
return WorkflowKey{
NamespaceID: namespaceID,
WorkflowID: workflowID,
RunID: runID,
}
}
func (k *WorkflowKey) GetNamespaceID() string {
}
return k.WorkflowID
}
return k.RunID
}
func (k *WorkflowKey) String() string {
}
d := metricDefinition{
name: name,
description: "",
unit: "",
}
for _, opt := range opts {
opt.apply(&d)
}
return d
}
return md.name
}
func (md metricDefinition) Unit() MetricUnit {
)
// WithTags creates a new MetricProvder with provided []Tag
// Gauge obtains a gauge for the given name.
return NoopGaugeMetricFunc
}
// Timer obtains a timer for the given name.
return NoopTimerMetricFunc
}
// Histogram obtains a histogram for the given name.
return NoopHistogramMetricFunc
}
func (*noopMetricsHandler) Stop(log.Logger) {}
var NoopCounterMetricFunc = CounterFunc(func(i int64, t ...Tag) {})
// NewVersionHistories create a new instance of VersionHistories.
func NewVersionHistories(versionHistory *historyspb.VersionHistory) *historyspb.VersionHistories {
version_histories.go
if versionHistory == nil {
panic("version history cannot be null")
}
CurrentVersionHistoryIndex: 0,
Histories: []*historyspb.VersionHistory{versionHistory},
}
}
// GetVersionHistory gets the VersionHistory according to index provided.
func GetVersionHistory(h *historyspb.VersionHistories, index int32) (*historyspb.VersionHistory, error) {
version_histories.go
if index < 0 || index >= int32(len(h.Histories)) {
return nil, serviceerror.NewInternal("version histories index is out of range.")
}
}
// GetCurrentVersionHistory gets the current VersionHistory.
func GetCurrentVersionHistory(h *historyspb.VersionHistories) (*historyspb.VersionHistory, error) {
version_histories.go
return GetVersionHistory(h, h.GetCurrentVersionHistoryIndex())
}
// IsCurrentVersionHistoryEmpty checks if the current VersionHistory is empty.
}
ret := make([]elastic.Sorter, 0, len(defaultSorterFields))
for _, item := range defaultSorterFields {
fs := elastic.NewFieldSort(item.name)
if item.desc {
fs.Desc()
}
if item.missing_first {
fs.Missing("_first")
} else {
fs.Missing("_last")
}
}
}()
)
return Key{
FireTime: DefaultFireTime,
TaskID: taskID,
}
}
return Key{
FireTime: fireTime,
TaskID: taskID,
}
}
func ValidateKey(key Key) error {
// NewDefaultHandoverTrackerFactory returns a factory that creates the default OSS HandoverTracker.
return func(params HandoverTrackerParams) HandoverTracker {
handoverNamespaces: make(map[namespace.Name]*namespaceHandOverInfo),
clusterMetadata: params.ClusterMetadata,
getMaxReplicationTaskID: params.GetMaxReplicationTaskID,
errorByStateFn: params.ErrorByStateFn,
notifyReplicationFn: params.NotifyReplicationFn,
logger: params.Logger,
}
}
}
)
dc := dynamicconfig.NewNoopCollection()
config := configs.NewConfig(dc, 1)
config.EnableActivityEagerExecution = dynamicconfig.GetBoolPropertyFnFilteredByNamespace(true)
config.NamespaceCacheRefreshInterval = dynamicconfig.GetDurationPropertyFn(time.Second)
config.ReplicationEnableUpdateWithNewTaskMerge = dynamicconfig.GetBoolPropertyFn(true)
config.EnableWorkflowIdReuseStartTimeValidation = dynamicconfig.GetBoolPropertyFnFilteredByNamespace(true)
config.EnableTransitionHistory = dynamicconfig.GetBoolPropertyFnFilteredByNamespace(true)
config.EnableChasm = dynamicconfig.GetBoolPropertyFnFilteredByNamespace(false)
return config
}
}
func accumulatedSkippedDuration(source *persistencespb.WorkflowExecutionInfo) time.Duration {
timeskipping.go
return source.GetTimeSkippingInfo().GetAccumulatedSkippedDuration().AsDuration()
}
// =============================================================================
}
return accumulatedSkippedDuration(ms.executionInfo)
}
// =============================================================================
ctx context.Context,
transactionPolicy historyi.TransactionPolicy,
if !ms.IsWorkflow() {
return false
}
case historyi.TransactionPolicyActive:
// 1. gate: only a running, time-skipping-enabled, idle workflow may skip time
// 4. task regeneration
return true
return false
default:
ms.logger.Error(fmt.Sprintf("closeTransactionHandleTimeSkipping: unknown transaction policy: %v", transactionPolicy),
var _ sqlplugin.Plugin = (*plugin)(nil)
sql.RegisterPlugin(PluginName, &plugin{
driver: &driver.PQDriver{},
queryConverter: &queryConverter{},
})
sql.RegisterPlugin(PluginNamePGX, &plugin{
driver: &driver.PGXDriver{},
queryConverter: &queryConverter{},
})
}
func (p *plugin) GetVisibilityQueryConverter() sqlplugin.VisibilityQueryConverter {
)
items := make([]string, len(fields))
for i, field := range fields {
items[i] = fmt.Sprintf("%s = excluded.%s", field, field)
}
return fmt.Sprintf(
// The WHERE clause ensures that no update occurs if the version is behind the saved version.
"ON CONFLICT (namespace_id, run_id) DO UPDATE SET %s WHERE executions_visibility.%s < EXCLUDED.%s",
strings.Join(items, ", "), sqlplugin.VersionColumnName, sqlplugin.VersionColumnName,
)
}
)
items := make([]string, len(fields))
for i, field := range fields {
items[i] = fmt.Sprintf("%s = excluded.%s", field, field)
}
return fmt.Sprintf(
// The WHERE clause ensures that no update occurs if the version is behind the saved version.
"ON CONFLICT (namespace_id, run_id) DO UPDATE SET %s WHERE executions_visibility.%s < EXCLUDED.%s",
strings.Join(items, ", "), sqlplugin.VersionColumnName, sqlplugin.VersionColumnName,
)
}
}
t := reflect.TypeFor[VisibilityRow]()
dbFields := make([]string, t.NumField())
for i := 0; i < t.NumField(); i++ {
f := t.Field(i)
dbFields[i] = f.Tag.Get("db")
if dbFields[i] == "" {
dbFields[i] = strcase.ToSnake(f.Name)
}
}
}
}
return &UnsafeSQLString{Val: val}
}
func NewColName(name string) *ColumnName {
}
func NewSAColumn(alias string, fieldName string, valueType enumspb.IndexedValueType) *SAColumn {
util.go
return &SAColumn{
Alias: alias,
FieldName: fieldName,
ValueType: valueType,
}
}
func NamespaceDivisionSAColumn() *SAColumn {
// NewMockProvider creates a new mock instance.
mock := &MockProvider{ctrl: ctrl}
mock.recorder = &MockProviderMockRecorder{mock}
return mock
}
// EXPECT returns an object that allows the caller to indicate expected use.
// NewMockManager creates a new mock instance.
mock := &MockManager{ctrl: ctrl}
mock.recorder = &MockManagerMockRecorder{mock}
return mock
}
// EXPECT returns an object that allows the caller to indicate expected use.
ms *MutableStateImpl,
metricsHandler metrics.Handler,
return &workflowTaskStateMachine{
ms: ms,
metricsHandler: metricsHandler,
}
}
func (m *workflowTaskStateMachine) ApplyWorkflowTaskScheduledEvent(
}
func (m *workflowTaskStateMachine) HasStartedWorkflowTask() bool {
workflow_task_state_machine.go
return m.ms.executionInfo.WorkflowTaskScheduledEventId != common.EmptyEventID &&
m.ms.executionInfo.WorkflowTaskStartedEventId != common.EmptyEventID
}
func (m *workflowTaskStateMachine) GetStartedWorkflowTask() *historyi.WorkflowTaskInfo {
// NewDisabledArchvialConfig returns an ArchivalConfig where archival is disabled for both the cluster and the namespace
return &archivalConfig{
staticClusterState: ArchivalDisabled,
dynamicClusterState: nil,
enableRead: nil,
namespaceDefaultState: enumspb.ARCHIVAL_STATE_DISABLED,
namespaceDefaultURI: "",
}
}
// NewEnabledArchivalConfig returns an ArchivalConfig where archival is enabled for both the cluster and the namespace
)
if v, ok := s[key]; ok {
if cvs, ok := v.([]ConstrainedValue); ok {
return cvs
return []ConstrainedValue{{Value: v}}
}
}
// NewNoopClient returns a Client that has no keys (a Collection using it will always return
// default values).
return StaticClient(nil)
}
// NewNoopCollection creates a new noop collection.
return NewCollection(NewNoopClient(), log.NewNoopLogger())
}
// maximum combined weight for concurrent access, capable of handling multiple priority levels.
// Most of the logic is taken directly from golang's semaphore.Weighted.
waitLists := make([]*list.List, NumPriorities)
for i := range waitLists {
waitLists[i] = list.New()
}
return &PrioritySemaphoreImpl{
size: n,
waitLists: waitLists,
}
}
)
out := make([]string, len(fields))
for i, field := range fields {
out[i] = prefix + field
}
return out
}
return strings.Join(appendPrefix(":", fields), ", ")
}
var keyCounter atomic.Int64
var zero S
var s ScopeType
switch any(zero).(type) {
case namespace.ID, namespace.Name:
s = ScopeNamespace
case global:
s = ScopeGlobal
default:
panic("testhooks: unknown scope type")
}
}
}
return c.id
}
return c.name
}
return c.cType
}
func (c Category) MarshalText() (text []byte, err error) {
}
return u.VisibilityTimestamp
}
u.VisibilityTimestamp = t
}
return CategoryTimer
}
func (u *WorkflowRunTimeoutTask) GetType() enumsspb.TaskType {
type noopChasmTree struct{}
return chasm.NodesMutation{}, nil
}
func (*noopChasmTree) Snapshot(*persistencespb.VersionedTransition) chasm.NodesSnapshot {
noop_chasm_tree.go
return chasm.NodesSnapshot{}
}
func (*noopChasmTree) PartitionedSnapshot(*persistencespb.VersionedTransition) (chasm.NodesSnapshot, *persistencespb.ChasmLocalState) {
}
return chasm.WorkflowArchetypeID
}
func (*noopChasmTree) EachPureTask(
// Serialize is a noop as Deserialize is not supported.
return nil, nil
}
return StateMachineType
}
return reg.RegisterMachine(stateMachineDefinition{})
}
// AddNextStateMachineTimerTask generates a state machine timer task if the first deadline doesn't have a task scheduled
// yet.
// filter out empty timer groups
timers := ms.GetExecutionInfo().StateMachineTimers
timers = slices.DeleteFunc(timers, func(timerGroup *persistencespb.StateMachineTimerGroup) bool {
return len(timerGroup.Infos) == 0
})
if len(timers) == 0 {
}
timerGroup := timers[0]
// NewMetadataMock returns a new MetadataMock which uses the provided controller to create a MockArchivalMetadata
// instance.
m := &metadataMock{
MockArchivalMetadata: NewMockArchivalMetadata(controller),
defaultHistoryConfig: NewDisabledArchvialConfig(),
defaultVisibilityConfig: NewDisabledArchvialConfig(),
}
return m
}
// MetadataMockRecorder is a wrapper around a ArchivalMetadata mock recorder.
type mutationFunc func(*Namespace)
f(ns)
}
// WithActiveCluster assigns the active cluster to a Namespace during a Clone
// WithGlobalFlag sets whether or not this Namespace is global.
return mutationFunc(
func(ns *Namespace) {
ns.replicationResolver.SetGlobalFlag(b)
})
}
)
items := make([]string, len(fields))
for i, field := range fields {
// This line is to ensure that no update occurs (for any column) if the version is behind the saved version.
items[i] = fmt.Sprintf("%v = IF(%v < VALUES(%v), VALUES(%v), %v)",
field, sqlplugin.VersionColumnName, sqlplugin.VersionColumnName, field, field)
}
return fmt.Sprintf("ON DUPLICATE KEY UPDATE %s", strings.Join(items, ", "))
}
)
s, ok := PriorityName[p]
if ok {
return s
}
return strconv.Itoa(int(p))
}
func getPriority(
class, subClass Priority,
return class | subClass
}
)
return &queryRegistryImpl{
buffered: make(map[string]query),
completed: make(map[string]query),
unblocked: make(map[string]query),
failed: make(map[string]query),
}
}
func (r *queryRegistryImpl) HasBufferedQuery() bool {
//
//nolint:revive // cognitive complexity to reduce after old code clean up
func GetEffectiveDeployment(versioningInfo *workflowpb.WorkflowExecutionVersioningInfo) *deploymentpb.Deployment {
util.go
if versioningInfo == nil {
return nil
} else if transition := versioningInfo.GetVersionTransition(); transition != nil {
if v := transition.GetDeploymentVersion(); v != nil { // v0.32
return worker_versioning.DeploymentFromExternalDeploymentVersion(v)
// 3. Behavior: this is returned when there is no override (most common case). Behavior is
// set based on the worker-sent deployment in the latest WFT completion.
func GetEffectiveVersioningBehavior(versioningInfo *workflowpb.WorkflowExecutionVersioningInfo) enumspb.VersioningBehavior {
util.go
if versioningInfo == nil {
return enumspb.VERSIONING_BEHAVIOR_UNSPECIFIED
} else if t := versioningInfo.GetVersionTransition(); t != nil {
return enumspb.VERSIONING_BEHAVIOR_AUTO_UPGRADE
} else if override := versioningInfo.GetVersioningOverride(); override != nil {
// For more information see the documentation for
// https://pkg.go.dev/google.golang.org/protobuf/proto#Equal
if that == nil {
return this == nil
}
switch t := that.(type) {
case *VersionHistoryItem:
that1 = t
case VersionHistoryItem:
that1 = &t
// The apply function is called after verifying the transition is possible but before setting the destination state,
// so it can inspect the current (source) state.
func NewTransition[S comparable, SM StateMachine[S], E any](src []S, dst S, apply func(SM, MutableContext, E) error) Transition[S, SM, E] {
statemachine.go
return Transition[S, SM, E]{
Sources: src,
Destination: dst,
apply: apply,
}
}
// Possible returns a boolean indicating whether the transition is possible for the current state.
)
if globalRegistry.queried.Load() {
panic("dynamicconfig.New*Setting must only be called from static initializers")
}
globalRegistry.settings = make(map[Key]GenericSetting)
}
if globalRegistry.settings[s.Key()] != nil {
// nolint:forbidigo // only called during static initialization
panic(fmt.Sprintf("duplicate registration of dynamic config key: %q", s.Key().String()))
}
}
)
// This must be called in init to avoid race conditions.
resolver.Register(&globalGrpcBuilder)
}
// Most code should not use this, this is only exposed for code that has to recognize and use a
}
return grpcResolverScheme
}
func (m *grpcBuilder) getServiceResolver(u *url.URL) (ServiceResolver, error) {
// NewTransition creates a new [Transition] from the given source states to a destination state for a given event.
// The apply function is called after verifying the transition is possible and setting the destination state.
func NewTransition[S comparable, SM StateMachine[S], E any](src []S, dst S, apply func(SM, E) (TransitionOutput, error)) Transition[S, SM, E] {
sm.go
return Transition[S, SM, E]{
Sources: src,
Destination: dst,
apply: apply,
}
}
// Possible returns a boolean indicating whether the transition is possible for the current state.
shardID int32,
clock int64,
return &clockspb.VectorClock{
ClusterId: clusterID,
ShardId: shardID,
Clock: clock,
}
}
func Comparable(
// Size returns the size of the object, in bytes, once serialized
return proto.Size(val)
}
// Equal returns whether two WorkflowExecutionInfo values are equivalent by recursively
// Size returns the size of the object, in bytes, once serialized
return proto.Size(val)
}
// Equal returns whether two WorkflowExecutionState values are equivalent by recursively
// NewRealTimeSource returns a timeSource that uses the real wall timeSource time.
return RealTimeSource{}
}
// Now returns the current time, with the location set to UTC.
return time.Now().UTC()
}
// Since returns the time elapsed since t
)
return &lazyLogger{
logger: logger,
tagFn: tagFn,
}
}
func (l *lazyLogger) Debug(msg string, tags ...tag.Tag) {
type WithDescription string
m.description = string(h)
}
// WithUnit sets the unit of a metric. See NewBytesHistogramDef for an example.
type WithUnit MetricUnit
m.unit = MetricUnit(h)
}
}
return Tag{Key: serviceName, Value: string(value)}
}
func ActionType(value string) Tag {
}
return Tag{Key: OperationTagName, Value: value}
}
func StringTag(key string, value string) Tag {
// UnixMilliseconds returns t as a Unix time, the number of milliseconds elapsed since January 1, 1970 UTC.
// It should be used for all CQL timestamp.
// Handling zero time separately because UnixNano is undefined for zero times.
if t.IsZero() {
return 0
}
if unixNano < 0 {
// Time is before January 1, 1970 UTC
return 0
}
}
)
RegisterTranslator(fixedTranslatorName, NewFixedAddressTranslatorPlugin())
}
type FixedAddressTranslatorPlugin struct {
}
return &FixedAddressTranslatorPlugin{}
}
// GetTranslator What gocql driver does is that it will connect to the first node in the list in configuration
)
return &serializerImpl{encodingType: encodingTypeFromEnv()}
}
func (t *serializerImpl) EncodingType() enumspb.EncodingType {
}
func (t *serializerImpl) HistoryBranchToBlob(info *persistencespb.HistoryBranch) (*commonpb.DataBlob, error) {
serializer.go
return encodeBlob(info, t.encodingType)
}
// NOTE: HistoryBranch does not have an encoding type; so we use the serializer's encoding type.
}
sql.RegisterPlugin(PluginName, &plugin{
queryConverter: &queryConverter{},
connPool: newConnPool(),
})
}
func (p *plugin) GetVisibilityQueryConverter() sqlplugin.VisibilityQueryConverter {
// NewVersionHistoryItem create a new instance of VersionHistoryItem.
func NewVersionHistoryItem(eventID int64, version int64) *historyspb.VersionHistoryItem {
version_history_item.go
if eventID < 0 || version < 0 {
panic(fmt.Sprintf("invalid version history item event ID: %v, version: %v", eventID, version))
}
return &historyspb.VersionHistoryItem{EventId: eventID, Version: version}
version_history_item.go
}
// CopyVersionHistoryItem create a new instance of VersionHistoryItem.
func CopyVersionHistoryItem(item *historyspb.VersionHistoryItem) *historyspb.VersionHistoryItem {
version_history_item.go
return NewVersionHistoryItem(item.EventId, item.Version)
}
// IsEqualVersionHistoryItem checks whether version history items are equal
func ConvertWeightsToDynamicConfigValue(
weights map[tasks.Priority]int,
weightsForDC := make(map[string]any)
for priority, weight := range weights {
weightsForDC[priority.String()] = weight
}
return weightsForDC
}
func NewTaskRefresher(
shard historyi.ShardContext,
return &TaskRefresherImpl{
shard: shard,
taskGeneratorProvider: GetTaskGeneratorProvider(),
}
}
func (r *TaskRefresherImpl) Refresh(
// NewMockAdminServiceClient creates a new mock instance.
func NewMockAdminServiceClient(ctrl *gomock.Controller) *MockAdminServiceClient {
service_grpc.pb.mock.go
mock := &MockAdminServiceClient{ctrl: ctrl}
mock.recorder = &MockAdminServiceClientMockRecorder{mock}
return mock
}
// EXPECT returns an object that allows the caller to indicate expected use.
// NewMockHistoryServiceClient creates a new mock instance.
func NewMockHistoryServiceClient(ctrl *gomock.Controller) *MockHistoryServiceClient {
service_grpc.pb.mock.go
mock := &MockHistoryServiceClient{ctrl: ctrl}
mock.recorder = &MockHistoryServiceClientMockRecorder{mock}
return mock
}
// EXPECT returns an object that allows the caller to indicate expected use.
// NewMockMatchingServiceClient creates a new mock instance.
func NewMockMatchingServiceClient(ctrl *gomock.Controller) *MockMatchingServiceClient {
service_grpc.pb.mock.go
mock := &MockMatchingServiceClient{ctrl: ctrl}
mock.recorder = &MockMatchingServiceClientMockRecorder{mock}
return mock
}
// EXPECT returns an object that allows the caller to indicate expected use.
}
return backoff.NewExponentialRetryPolicy(cfg.InitialInterval).
WithMaximumInterval(cfg.MaxInterval).
WithExpirationInterval(backoff.NoInterval)
}
var defaultRetryPolicyConfig = RetryPolicyConfig{
// NewNexusEndpointProcessor creates a new NexusEndpointProcessor.
return &NexusEndpointProcessor{
serviceProcessors: make(map[string]*NexusServiceProcessor),
}
}
// RegisterServiceProcessor adds a service-level processor to the endpoint keyed by its name.
// NewMockFactory creates a new mock instance.
mock := &MockFactory{ctrl: ctrl}
mock.recorder = &MockFactoryMockRecorder{mock}
return mock
}
// EXPECT returns an object that allows the caller to indicate expected use.
// NewMockArchivalMetadata creates a new mock instance.
func NewMockArchivalMetadata(ctrl *gomock.Controller) *MockArchivalMetadata {
archival_metadata_mock.go
mock := &MockArchivalMetadata{ctrl: ctrl}
mock.recorder = &MockArchivalMetadataMockRecorder{mock}
return mock
}
// EXPECT returns an object that allows the caller to indicate expected use.
// NewMockArchiverProvider creates a new mock instance.
mock := &MockArchiverProvider{ctrl: ctrl}
mock.recorder = &MockArchiverProviderMockRecorder{mock}
return mock
}
// EXPECT returns an object that allows the caller to indicate expected use.
// register adds a metric definition to the list of pending metric definitions. This method is thread-safe.
c.Lock()
defer c.Unlock()
c.definitions = append(c.definitions, d)
}
// buildCatalog builds a catalog from the list of pending metric definitions. It is safe to call this method multiple
)
// Error can be safely ignored here becase string always can be converted.
ps, _ := defaultDataConverter.ToPayloads(str)
return ps
}
func EncodeInt(i int) *commonpb.Payloads {
// NewMockNamespaceReplicationQueue creates a new mock instance.
func NewMockNamespaceReplicationQueue(ctrl *gomock.Controller) *MockNamespaceReplicationQueue {
namespace_replication_queue_mock.go
mock := &MockNamespaceReplicationQueue{ctrl: ctrl}
mock.recorder = &MockNamespaceReplicationQueueMockRecorder{mock}
return mock
}
// EXPECT returns an object that allows the caller to indicate expected use.
var _ sqlplugin.Plugin = (*plugin)(nil)
sql.RegisterPlugin(PluginName, &plugin{
queryConverter: &queryConverter{},
})
}
func (p *plugin) GetVisibilityQueryConverter() sqlplugin.VisibilityQueryConverter {
}
return &connPool{
pool: make(map[string]entry),
}
}
// Allocate allocates the shared database in the pool or returns already exists instance with the same DSN. If instance
// NewMockVisibilityManager creates a new mock instance.
func NewMockVisibilityManager(ctrl *gomock.Controller) *MockVisibilityManager {
visibility_manager_mock.go
mock := &MockVisibilityManager{ctrl: ctrl}
mock.recorder = &MockVisibilityManagerMockRecorder{mock}
return mock
}
// EXPECT returns an object that allows the caller to indicate expected use.
// NewMockClient creates a new mock instance.
mock := &MockClient{ctrl: ctrl}
mock.recorder = &MockClientMockRecorder{mock}
return mock
}
// EXPECT returns an object that allows the caller to indicate expected use.
// NewMockClientFactory creates a new mock instance.
mock := &MockClientFactory{ctrl: ctrl}
mock.recorder = &MockClientFactoryMockRecorder{mock}
return mock
}
// EXPECT returns an object that allows the caller to indicate expected use.
// NewMockMapperProvider creates a new mock instance.
mock := &MockMapperProvider{ctrl: ctrl}
mock.recorder = &MockMapperProviderMockRecorder{mock}
return mock
}
// EXPECT returns an object that allows the caller to indicate expected use.
// NewMockWorkflowServiceClient creates a new mock instance.
func NewMockWorkflowServiceClient(ctrl *gomock.Controller) *MockWorkflowServiceClient {
service_grpc.pb.mock.go
mock := &MockWorkflowServiceClient{ctrl: ctrl}
mock.recorder = &MockWorkflowServiceClientMockRecorder{mock}
return mock
}
// EXPECT returns an object that allows the caller to indicate expected use.
}
tagsToFilter := make(map[string]map[string]struct{})
for key, val := range cfg.ExcludeTags {
exclusions := make(map[string]struct{})
for _, val := range val {
)
}
t, err := time.Parse(time.RFC3339, "1000-01-01T00:00:00Z")
if err != nil {
return time.Unix(0, 0).UTC()
}
}
}
t, err := time.Parse(time.RFC3339, "1000-01-01T00:00:00Z")
if err != nil {
return time.Unix(0, 0).UTC()
}
}
}
t, err := time.Parse(time.RFC3339, "1000-01-01T00:00:00Z")
if err != nil {
return time.Unix(0, 0).UTC()
}
}
// CreatePersistenceClientRetryPolicy creates a retry policy for calls to persistence
return backoff.NewExponentialRetryPolicy(persistenceClientRetryInitialInterval).
WithMaximumAttempts(persistenceClientRetryMaxAttempts)
}
// CreateFrontendClientRetryPolicy creates a retry policy for calls to frontend service
// InverseMap creates the inverse map, ie., for a key-value map, it builds the value-key map.
if m == nil {
return nil
}
invm := make(map[V]K, len(m))
for k, v := range m {
// Size returns the size of the object, in bytes, once serialized
return proto.Size(val)
}
// Equal returns whether two Predicate values are equivalent by recursively
// tasks within the CHASM framework.
// The format of the returned FQN is: "libName.name"
return libName + "." + name
}
// The generated ID is used to uniquely identify components and tasks within the CHASM framework. The same FQN will
// always produce the same ID.
return farm.Fingerprint32([]byte(fqn))
}
// hasBusinessIDAlias returns true if the component has a businessID alias configured
)
return Key{handle: unique.Make(strings.ToLower(s))}
}
func (k Key) String() string {
// NewNoopLogger return a noopLogger
return &noopLogger{}
}
func (n *noopLogger) Debug(string, ...tag.Tag) {}
// NewHostInfoFromAddress creates a new HostInfo instance from a socket address.
return hostAddress(address)
}
// hostAddress is a HostInfo implementation that uses a string as the address and identity.
}
return defaultDataConverter.ToPayload(value)
}
func Decode(p *commonpb.Payload, valuePtr any) error {
// RegisterPlugin adds an auth plugin to the plugin registry
// it is only safe to use from a package init function
translators[name] = plugin
}
func LookupTranslator(name string) (TranslatorPlugin, error) {
baseAPI string,
taskCategory tasks.Category,
return baseAPI + taskCategory.Name()
}
// RegisterPlugin will register a SQL plugin
if _, ok := supportedPlugins[pluginName]; ok {
panic("plugin " + pluginName + " already registered")
}
}
}
return CategoryVisibility
}
func (t *StartExecutionVisibilityTask) GetType() enumsspb.TaskType {
}
func managerProvider[T persistence.Closeable](newManagerFn func(Factory) (T, error)) func(Factory, fx.Lifecycle) (T, error) {
fx.go
return func(f Factory, lc fx.Lifecycle) (T, error) {
manager, err := newManagerFn(f) // passing receiver (Factory) as first argument.
if err != nil {
)
func newNoopMovingWindowAverage() *noopMovingWindowAverage { return &noopMovingWindowAverage{} }
noop_moving_window_average.go
func (a *noopMovingWindowAverage) Record(_ int64) {}
)
func newNoopSignalAggregator() *noopSignalAggregator { return &noopSignalAggregator{} }
noop_health_signal_aggregator.go
func (a *noopSignalAggregator) Start() {}