Atlas › Test
Exact test identity: go.temporal.io/server/service/history/shard/TestShardControllerSuite/TestShardLingerTimeout
go.temporal.io/server/service/history/shardTestShardControllerSuite/TestShardLingerTimeoutTestShardLingerTimeoutExpand a file to inspect source; the > gutter marks covered lines.
}
// constant from initialization, no need for locks
return s.shardID
}
func (s *ContextImpl) GetRangeID() int64 {
func (s *ContextImpl) GetEngine(
ctx context.Context,
return s.engineFuture.Get(ctx)
}
func (s *ContextImpl) AssertOwnership(
ctx context.Context,
if err := s.ioSemaphoreAcquire(ctx); err != nil {
}
s.wLock()
// timeout check should be done within the shard lock, in case of shard lock contention
ctx, cancel, err := s.newDetachedContext(ctx)
if err != nil {
s.wUnlock()
return err
}
if err := s.errorByState(); err != nil {
s.wUnlock()
return err
}
ShardID: s.shardID,
RangeID: s.getRangeIDLocked(),
}
s.wUnlock()
err = s.persistenceShardManager.AssertShardOwnership(ctx, request)
return s.handleWriteError(request.RangeID, err)
}
}
// constant from initialization, no need for locks
return s.config
}
func (s *ContextImpl) GetEventsCache() events.Cache {
}
// constant from initialization, no need for locks
return s.contextTaggedLogger
}
func (s *ContextImpl) GetThrottledLogger() log.Logger {
}
return s.shardInfo.GetRangeId()
}
s.stateLock.Lock()
defer s.stateLock.Unlock()
switch s.state {
case contextStateInitialized, contextStateAcquiring:
return ErrShardStatusUnknown
return nil
case contextStateStopping, contextStateStopped:
return s.newShardClosedErrorWithShardID()
}
// We must drain all in-flight requests before updating the rangeID.
// This is because requests are conditioned on rangeID, if rangeID
// is updated before draining them, those requests could fail.
// This also means renew rangeID will be the only in-flight request
// when it's issued, so it doesn't matter if semaphore is acquired or not
// before calling this method.
s.taskKeyManager.drainTaskRequests()
updatedShardInfo := trimShardInfo(s.config, s.clusterMetadata.GetAllClusterInfo(), s.copyShardInfo(s.shardInfo))
updatedShardInfo.RangeId++
if isStealing {
updatedShardInfo.StolenSinceRenew++
}
defer cancel()
previousRangeID := s.getRangeIDLocked()
err := s.persistenceShardManager.UpdateShard(ctx, &persistence.UpdateShardRequest{
ShardInfo: updatedShardInfo,
PreviousRangeID: previousRangeID,
})
if err != nil {
// Failure in updating shard to grab new RangeID
s.contextTaggedLogger.Error("Persistent store operation failure",
// Range is successfully updated in cassandra now update shard context to reflect new range
tag.ShardRangeID(updatedShardInfo.RangeId),
tag.PreviousShardRangeID(s.shardInfo.RangeId),
)
s.shardInfo = trimShardInfo(s.config, s.clusterMetadata.GetAllClusterInfo(), s.copyShardInfo(updatedShardInfo))
s.taskKeyManager.setRangeID(s.shardInfo.RangeId)
return nil
}
timer := time.NewTimer(queueMetricUpdateInterval)
defer timer.Stop()
done := s.lifecycleCtx.Done()
for {
select {
return
case <-timer.C:
s.emitShardInfoMetricsLogs()
}
if cluster != s.GetClusterMetadata().GetCurrentClusterName() {
s.wLock()
defer s.wUnlock()
return s.getOrUpdateRemoteClusterInfoLocked(cluster).CurrentTime
}
}
s.rLock()
defer s.rUnlock()
return s.lastUpdated
}
func (s *ContextImpl) handleReadError(err error) error {
requestRangeID int64,
err error,
s.wLock()
defer s.wUnlock()
return s.handleWriteErrorLocked(requestRangeID, err)
}
func (s *ContextImpl) handleWriteErrorLocked(
requestRangeID int64,
err error,
if requestRangeID != s.getRangeIDLocked() {
return err
}
return err
}
// Persistence success: update max read level
return nil
case *persistence.AppendHistoryTimeoutError:
}
func (s *ContextImpl) maybeRecordShardAcquisitionLatency(ownershipChanged bool) {
context_impl.go
if ownershipChanged {
metrics.ShardContextAcquisitionLatency.With(s.GetMetricsHandler()).
Record(s.GetCurrentTime(s.GetClusterMetadata().GetCurrentClusterName()).Sub(s.getLastUpdatedTime()),
metrics.OperationTag(metrics.ShardInfoScope),
)
}
}
s.contextTaggedLogger.Info("", tag.LifeCycleStarting, tag.ComponentShardEngine)
engine := s.engineFactory.CreateEngine(s)
engine.Start()
s.contextTaggedLogger.Info("", tag.LifeCycleStarted, tag.ComponentShardEngine)
return engine
}
// start should only be called by the controller.
_ = s.transition(contextRequestAcquire{})
}
func (s *ContextImpl) UnloadForOwnershipLost() {
// 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)
context_impl.go
engine.Stop()
s.contextTaggedLogger.Info("", tag.LifeCycleStopped, tag.ComponentShardEngine)
}
// Run finalizer to cleanup any of the shard's associated resources that are registered.
}
}
s.stateLock.Lock()
defer s.stateLock.Unlock()
return s.state < contextStateStopping
}
func (s *ContextImpl) GetLifecycleContext() context.Context {
}
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(
ctx context.Context,
priority := locks.PriorityHigh
callerInfo := headers.GetCallerInfo(ctx)
if callerInfo.CallerType == headers.CallerTypePreemptable {
priority = locks.PriorityLow
}
handler := s.metricsHandler.WithTags(metrics.OperationTag(metrics.ShardInfoScope), metrics.PriorityTag(priority))
context_impl.go
metrics.SemaphoreRequests.With(handler).Record(1)
startTime := time.Now().UTC()
defer func() {
metrics.SemaphoreLatency.With(handler).Record(time.Since(startTime))
if retErr != nil {
}
}()
}
s.ioSemaphore.Release(1)
}
/* 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.contextTaggedLogger.Info("", tag.LifeCycleStarted, tag.ComponentShardContext)
go s.acquireShard()
}
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()
}
switch request := request.(type) {
case contextRequestAcquire:
setStateAcquiring()
return nil
case contextRequestStop:
setStateStopping(request)
return nil
}
switch request := request.(type) {
case contextRequestAcquire:
return nil // nothing to do, already acquiring
s.state = contextStateAcquired
if request.engine != nil {
// Acquiring, so that other code (i.e. FinishStop) can know that after a state
// transition to Stopping/Stopped, engineFuture cannot be Set.
if s.engineFuture.Ready() {
// defensive check, this should never happen
s.contextTaggedLogger.Warn("transition to acquired with engine set twice")
return errInvalidTransition
}
}
// we should either have an engine from a previous transition, or set one now
s.contextTaggedLogger.Warn("transition to acquired but no engine set")
}
case contextRequestLost:
return nil // nothing to do, already acquiring
return nil
}
switch request := request.(type) {
case contextRequestAcquire:
return nil // nothing to do, already acquired
setStateStopping(request)
return nil
setStateStopped()
return nil
}
case contextStateStopping:
// notifyQueueProcessor sends notification to all queue processors for triggering a load
// NOTE: this method assumes engineFuture is already in a ready state.
// use a cancelled ctx so the method won't be blocked if engineFuture is not ready
cancelledCtx, cancel := context.WithCancel(context.Background())
cancel()
// we will get the engine when the Future is ready
engine, err := s.engineFuture.Get(cancelledCtx)
if err != nil {
s.contextTaggedLogger.Warn("tried to notify queue processor when engine is not ready")
return
}
fakeTasks := make(map[tasks.Category][]tasks.Task)
for _, category := range s.taskCategoryRegistry.GetCategories() {
fakeTasks[category] = []tasks.Task{tasks.NewFakeTask(definition.WorkflowKey{}, category, now)}
}
}
s.wLock()
if s.errorByState() != nil {
// if not in acquired state, this function will be called again
// later when shard is re-acquired.
}
s.handoverTracker.ResolvePendingTaskIDs(maxReplicationTaskID)
s.wUnlock()
s.notifyReplicationQueueProcessor(maxReplicationTaskID)
}
return s.taskKeyManager.getExclusiveReaderHighWatermark(tasks.CategoryReplication).TaskID - 1
}
// Replication ack level won't exceed the max taskID it received via task notification.
// Since here we want it's ack level to advance to at least the input taskID, we need to
// trigger an fake notification.
cancelledCtx, cancel := context.WithCancel(context.Background())
cancel()
engine, err := s.engineFuture.Get(cancelledCtx)
if err != nil {
s.contextTaggedLogger.Warn("tried to notify replication queue processor when engine is not ready")
return
}
fakeReplicationTask := tasks.NewFakeTask(definition.WorkflowKey{}, tasks.CategoryReplication, tasks.MinimumKey.FireTime)
context_impl.go
fakeReplicationTask.SetTaskID(taskID)
engine.NotifyNewTasks(map[tasks.Category][]tasks.Task{
tasks.CategoryReplication: {fakeReplicationTask},
})
}
// Only have to do this once, we can just re-acquire the rangeid lock after that
s.rLock()
if s.shardInfo != nil {
s.rUnlock()
return nil
}
// We don't have any shardInfo yet, load it (outside of context rwlock)
ctx, cancel := s.newIOContext()
defer cancel()
resp, err := s.persistenceShardManager.GetOrCreateShard(ctx, &persistence.GetOrCreateShardRequest{
ShardID: s.shardID,
LifecycleContext: s.lifecycleCtx,
})
if err != nil {
s.contextTaggedLogger.Error("Failed to load shard", tag.Error(err))
return err
}
shardInfo := trimShardInfo(s.config, s.clusterMetadata.GetAllClusterInfo(), s.copyShardInfo(resp.ShardInfo))
shardInfo.Owner = s.owner
// initialize the cluster current time to be the same as ack level
remoteClusterInfos := make(map[string]*remoteClusterInfo)
var taskMinScheduledTime time.Time
currentClusterName := s.GetClusterMetadata().GetCurrentClusterName()
taskCategories := s.taskCategoryRegistry.GetCategories()
for clusterName, info := range s.GetClusterMetadata().GetAllClusterInfo() {
if !info.Enabled {
continue
}
for categoryID, queueState := range shardInfo.QueueStates {
if !ok || category.Type() != tasks.CategoryTypeScheduled {
continue
}
exclusiveMaxReadTime = util.MaxTime(exclusiveMaxReadTime, timestamp.TimeValue(queueState.ExclusiveReaderHighWatermark.FireTime))
context_impl.go
}
// Once we validate the rest of the code can worker correctly with higher precision, the code should simply be
// taskMinScheduledTime = util.MaxTime(taskMinScheduledTime, maxReadTime)
taskMinScheduledTime,
exclusiveMaxReadTime.Add(common.ScheduledTaskMinPrecision).Truncate(common.ScheduledTaskMinPrecision),
)
if clusterName != currentClusterName {
remoteClusterInfos[clusterName] = &remoteClusterInfo{
CurrentTime: exclusiveMaxReadTime,
}
defer s.wUnlock()
s.shardInfo = shardInfo
s.remoteClusterInfos = remoteClusterInfos
s.taskKeyManager.setTaskMinScheduledTime(taskMinScheduledTime)
return nil
}
}
// This is called in two contexts: initially acquiring the rangeid lock, and trying to
// re-acquire it after a persistence error. In both cases, we retry the acquire operation
// (renewRangeLocked) for 5 minutes. Each individual attempt uses shardIOTimeout (by default, 5s) as
// the timeout. This lets us handle a few minutes of persistence unavailability without
// dropping and reloading the whole shard context, which is relatively expensive (includes
// caches that would have to be refilled, etc.).
//
// We stop retrying on any of:
// 1. We succeed in acquiring the rangeid lock.
// 2. We get ShardOwnershipLostError or lifecycleCtx ended.
// 3. The state changes to Stopping or Stopped.
//
// If the shard controller sees that service resolver has assigned ownership to someone
// else, it will call FinishStop, which will trigger case 3 above, and also cancel
// lifecycleCtx. The persistence operations called here use lifecycleCtx as their context,
// so if we were blocked in any of them, they should return immediately with a context
// canceled error.
policy := s.acquireShardRetryPolicy
if policy == nil {
policy = backoff.NewExponentialRetryPolicy(1 * time.Second).WithExpirationInterval(5 * time.Minute)
context_impl.go
}
// Remember this value across attempts
op := func() error {
if !s.IsValid() {
return s.newShardClosedErrorWithShardID()
}
// Initial load of shard metadata
if err != nil {
return err
}
// in-flight requests before making the call. So it's guaranteed that the renew rangeID
// UpdateShard call is the only one in flight.
err = s.renewRangeLocked(true)
s.wUnlock()
if err != nil {
return err
}
// The first time we get the shard, we have to create the engine
var engine historyi.Engine
if !s.engineFuture.Ready() {
engine = s.createEngine()
}
// NOTE: engine is created & started before setting shard state to acquired.
// -> information for handover namespace is recorded before shard can servce traffic
// -> upon shard reload, no history api or task can go through for ns in handover state
if err != nil {
if engine != nil {
// We tried to set the engine but the context was already stopped
// we know engineFuture must be ready here, and we can notify queue processor
// to trigger a load as queue max level can be updated to a newer value
// This runs until the lifecycleCtx is cancelled, so we only need to start it once
s.queueMetricEmitter.Do(func() {
go s.monitorQueueMetrics()
})
return nil
}
// keep retrying except ShardOwnershipLostError or lifecycle context ended
defer func() {
s.contextTaggedLogger.Error(
return true
}
if err != nil {
// We got an non-retryable error, e.g. ShardOwnershipLostError
s.contextTaggedLogger.Error("Couldn't acquire shard", tag.Error(err))
endpointRegistry chasm.EndpointRegistry,
handoverTrackerFactory HandoverTrackerFactory,
hostIdentity := hostInfoProvider.HostInfo().Identity()
sequenceID := atomic.AddInt64(&shardContextSequenceID, 1)
lifecycleCtx, lifecycleCancel := context.WithCancel(context.Background())
ioConcurrency := historyConfig.ShardIOConcurrency()
if ioConcurrency != 1 && persistenceConfig.DataStores[persistenceConfig.DefaultStore].Cassandra != nil {
throttledLogger.Warn(
fmt.Sprintf("Cassandra persistence implementation only supports %v == 1", dynamicconfig.ShardIOConcurrency),
}
taggedLogger := log.With(logger, tag.ShardID(shardID), tag.Address(hostIdentity))
context_impl.go
shardContext := &ContextImpl{
state: contextStateInitialized,
shardID: shardID,
owner: fmt.Sprintf("%s-%v-%v", hostIdentity, sequenceID, uuid.NewString()),
stringRepr: fmt.Sprintf("Shard(%d)", shardID),
executionManager: persistenceExecutionManager,
metricsHandler: metricsHandler,
eventLogger: eventLogger,
closeCallback: closeCallback,
config: historyConfig,
finalizer: finalizer.New(taggedLogger, metricsHandler),
contextTaggedLogger: taggedLogger,
throttledLogger: log.With(throttledLogger, tag.ShardID(shardID), tag.Address(hostIdentity)),
engineFactory: factory,
persistenceShardManager: persistenceShardManager,
clientBean: clientBean,
historyClient: historyClient,
payloadSerializer: payloadSerializer,
timeSource: timeSource,
namespaceRegistry: namespaceRegistry,
saProvider: saProvider,
saMapperProvider: saMapperProvider,
clusterMetadata: clusterMetadata,
archivalMetadata: archivalMetadata,
hostInfoProvider: hostInfoProvider,
taskCategoryRegistry: taskCategoryRegistry,
lifecycleCtx: lifecycleCtx,
lifecycleCancel: lifecycleCancel,
engineFuture: future.NewFuture[historyi.Engine](),
queueMetricEmitter: sync.Once{},
ioSemaphore: locks.NewPrioritySemaphore(ioConcurrency),
stateMachineRegistry: stateMachineRegistry,
chasmRegistry: chasmRegistry,
chasmWorkflowRegistry: chasmWorkflowRegistry,
endpointRegistry: endpointRegistry,
businessIDRateLimiters: cache.New(
historyConfig.BusinessIDReuseLimiterCacheSize(),
&cache.Options{TTL: historyConfig.BusinessIDReuseLimiterCacheTTL()},
),
}
shardContext.taskKeyManager = newTaskKeyManager(
shardContext.taskCategoryRegistry,
timeSource,
historyConfig,
shardContext.GetLogger(),
func() error {
return shardContext.renewRangeLocked(false)
},
)
ClusterMetadata: clusterMetadata,
GetMaxReplicationTaskID: shardContext.getMaxReplicationTaskID,
ErrorByStateFn: shardContext.errorByState,
NotifyReplicationFn: shardContext.notifyReplicationQueueProcessor,
NamespaceRegistry: namespaceRegistry,
Logger: taggedLogger,
})
if shardContext.GetConfig().EnableHostLevelEventsCache() {
shardContext.eventsCache = eventsCache
shardContext.eventsCache = events.NewShardLevelEventsCache(
shardContext.executionManager,
shardContext.config,
shardContext.metricsHandler,
shardContext.contextTaggedLogger,
false,
)
}
shardContext.initLastUpdatesTime()
return shardContext, nil
}
// We need to set lastUpdate time to "now" - "wait between shard updates time" + "first update interval".
// This is done to make sure that first shard update` will happen around "first update interval" after "now".
// The idea is to allow queue to persist even in the case of (relativly) constantly
// moving shards between hosts.
// Note: it still may prevent queue from progressing if shard moving rate is too high
lastUpdated := s.timeSource.Now()
lastUpdated = lastUpdated.Add(-1 * s.config.ShardUpdateMinInterval())
lastUpdated = lastUpdated.Add(s.config.ShardFirstUpdateInterval())
s.lastUpdated = lastUpdated
}
// TODO: why do we need a deep copy here?
func (s *ContextImpl) copyShardInfo(shardInfo *persistencespb.ShardInfo) *persistencespb.ShardInfo {
context_impl.go
// need to ser/de to make a deep copy of queue state
queueStates := make(map[int32]*persistencespb.QueueState, len(shardInfo.QueueStates))
for k, v := range shardInfo.QueueStates {
queueState, _ := s.payloadSerializer.QueueStateFromBlob(blob)
queueStates[k] = queueState
}
ShardId: shardInfo.ShardId,
Owner: shardInfo.Owner,
RangeId: shardInfo.RangeId,
StolenSinceRenew: shardInfo.StolenSinceRenew,
ReplicationDlqAckLevel: maps.Clone(shardInfo.ReplicationDlqAckLevel),
UpdateTime: shardInfo.UpdateTime,
QueueStates: queueStates,
}
}
}
return s.metricsHandler
}
return s.timeSource
}
func (s *ContextImpl) GetNamespaceRegistry() namespace.Registry {
}
return s.clusterMetadata
}
func (s *ContextImpl) GetArchivalMetadata() archiver.ArchivalMetadata {
func (s *ContextImpl) newDetachedContext(
ctx context.Context,
if err := ctx.Err(); err != nil {
return nil, nil, err
}
var cancel context.CancelFunc
deadline, ok := ctx.Deadline()
if ok {
detachedContext, cancel = context.WithTimeout(detachedContext, timeout)
cancel = func() {}
}
}
ctx, cancel := context.WithTimeout(s.lifecycleCtx, s.config.ShardIOTimeout())
ctx = headers.SetCallerInfo(ctx, headers.SystemBackgroundHighCallerInfo)
return ctx, cancel
}
// newShardClosedErrorWithShardID when shard is closed and a req cannot be processed
allClusterInfo map[string]cluster.ClusterInformation,
shardInfo *persistencespb.ShardInfo,
if shardInfo.QueueStates != nil && shardInfo.QueueStates[int32(tasks.CategoryIDReplication)] != nil {
for readerID := range shardInfo.QueueStates[int32(tasks.CategoryIDReplication)].ReaderStates {
context_impl.go
clusterID, _ := ReplicationReaderIDToClusterShardID(readerID)
_, clusterInfo, found := clusterNameInfoFromClusterID(allClusterInfo, clusterID)
if !found || !cluster.IsReplicationEnabledForCluster(clusterInfo, cfg.EnableSeparateReplicationEnableFlag()) {
delete(shardInfo.QueueStates[int32(tasks.CategoryIDReplication)].ReaderStates, readerID)
}
}
if len(shardInfo.QueueStates[int32(tasks.CategoryIDReplication)].ReaderStates) == 0 {
context_impl.go
delete(shardInfo.QueueStates, int32(tasks.CategoryIDReplication))
}
}
}
allClusterInfo map[string]cluster.ClusterInformation,
clusterID int64,
for name, info := range allClusterInfo {
if info.InitialFailoverVersion == clusterID {
return name, info, true
}
}
}
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 {
}
}
// 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 {
prec := []Constraints{{Namespace: namespace}, {}}
return matchAndConvert(
}
func GetTypedPropertyFnFilteredByNamespace[T any](value T) TypedPropertyFnWithNamespaceFilter[T] {
setting_gen.go
return func(namespace string) T {
return value
}
// 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(
hostInfoProvider membership.HostInfoProvider,
contextFactory ContextFactory,
hostIdentity := hostInfoProvider.HostInfo().Identity()
contextTaggedLogger := log.With(logger, tag.ComponentShardController, tag.Address(hostIdentity))
taggedMetricsHandler := metricsHandler.WithTags(metrics.OperationTag(metrics.HistoryShardControllerScope))
ownership := newOwnership(
config,
historyServiceResolver,
hostInfoProvider,
contextTaggedLogger,
taggedMetricsHandler,
)
c := &ControllerImpl{
config: config,
contextFactory: contextFactory,
contextTaggedLogger: contextTaggedLogger,
historyShards: make(map[int32]historyi.ControllableContext),
hostInfoProvider: hostInfoProvider,
ownership: ownership,
taggedMetricsHandler: taggedMetricsHandler,
shardCountSubscriptions: map[*shardCountSubscription]struct{}{},
initialShardsAcquired: future.NewFuture[struct{}](),
}
c.lingerState.shards = make(map[historyi.ControllableContext]struct{})
return c
}
func (c *ControllerImpl) Start() {
func (c *ControllerImpl) GetShardByID(
shardID int32,
startTime := time.Now().UTC()
defer func() {
metrics.GetEngineForShardLatency.With(c.taggedMetricsHandler).Record(time.Since(startTime))
}()
}
}
c.RLock()
defer c.RUnlock()
ids := make([]int32, 0, len(c.historyShards))
for id := range c.historyShards {
}
}
func (c *ControllerImpl) shardRemoveAndStop(shard historyi.ControllableContext) {
controller_impl.go
startTime := time.Now().UTC()
defer func() {
metrics.RemoveEngineForShardLatency.With(c.taggedMetricsHandler).Record(time.Since(startTime))
}()
_ = c.removeShard(shard.GetShardID(), shard)
// Whether shard was in the shards map or not, in both cases we should stop it.
shard.FinishStop()
}
// if necessary. If a shard context is created, it will initialize in the background.
// This function won't block on rangeid lease acquisition.
func (c *ControllerImpl) getOrCreateShardContext(shardID int32) (historyi.ControllableContext, error) {
controller_impl.go
if err := c.validateShardId(shardID); err != nil {
return nil, err
}
if shard, ok := c.historyShards[shardID]; ok {
c.RUnlock()
return shard, nil
}
// if shard not valid then proceed to create a new one
}
c.Lock()
defer c.Unlock()
// Check again with exclusive lock
if shard, ok := c.historyShards[shardID]; ok {
if shard.IsValid() {
return shard, nil
}
return nil, err
}
hostInfo := c.hostInfoProvider.HostInfo()
return nil, fmt.Errorf("ControllerImpl for host '%v' shutting down", hostInfo.Identity())
}
shard, err := c.contextFactory.CreateContext(shardID, c.shardRemoveAndStop)
controller_impl.go
if err != nil {
return nil, err
}
metrics.ShardContextCreatedCounter.With(c.taggedMetricsHandler).Record(1)
c.contextTaggedLogger.Info("", numShardsTag(len(c.historyShards)))
return shard, nil
}
func (c *ControllerImpl) removeShard(shardID int32, expected historyi.ControllableContext) historyi.ControllableContext {
controller_impl.go
c.Lock()
defer c.Unlock()
return c.removeShardLocked(shardID, expected)
}
func (c *ControllerImpl) removeShardLocked(shardID int32, expected historyi.ControllableContext) historyi.ControllableContext {
controller_impl.go
current, ok := c.historyShards[shardID]
if !ok {
return nil
}
// the shard comparison is a defensive check to make sure we are deleting
// what we intend to delete.
}
c.contextTaggedLogger.Info("", numShardsTag(len(c.historyShards)))
metrics.ShardContextRemovedCounter.With(c.taggedMetricsHandler).Record(1)
return current
}
// history instance can continue to process requests for the shard until the
// new owner actually acquires the shard.
func (c *ControllerImpl) shardLingerThenClose(ctx context.Context, shardID int32) {
controller_impl.go
c.RLock()
shard, ok := c.historyShards[shardID]
c.RUnlock()
if !ok {
return
}
// could be lingering on a shard that this instance should own, but this
// instance's acquireShards concurrency slots are filled with lingering shards.
return
}
defer c.endLinger(shard)
c.doLinger(ctx, shard)
}()
}
func (c *ControllerImpl) beginLinger(shard historyi.ControllableContext) bool {
controller_impl.go
c.lingerState.Lock()
defer c.lingerState.Unlock()
if _, ok := c.lingerState.shards[shard]; ok {
return false
}
return true
}
c.lingerState.Lock()
defer c.lingerState.Unlock()
delete(c.lingerState.shards, shard)
}
func (c *ControllerImpl) doLinger(ctx context.Context, shard historyi.ControllableContext) {
controller_impl.go
startTime := time.Now()
// Enforce a max limit to ensure we close the shard in a reasonable time,
// and to indirectly limit the number of lingering shards.
timeLimit := min(c.config.ShardLingerTimeLimit(), shardLingerMaxTimeLimit)
ctx, cancel := context.WithTimeout(ctx, timeLimit)
defer cancel()
qps := c.config.ShardLingerOwnershipCheckQPS()
// The limiter must be configured with burst>=1. With burst=1,
// the first call to Wait() won't be delayed.
limiter := rate.NewLimiter(rate.Limit(qps), 1)
for {
if !shard.IsValid() {
metrics.ShardLingerSuccess.With(c.taggedMetricsHandler).Record(time.Since(startTime))
break
}
tag.ShardID(shard.GetShardID()),
tag.Duration("duration", time.Since(startTime)),
)
metrics.ShardLingerTimeouts.With(c.taggedMetricsHandler).Record(1)
break
}
// If this AssertOwnership or any other request on the shard receives
// a shard ownership lost error, the shard will be marked as invalid.
}
}
metrics.AcquireShardsCounter.With(c.taggedMetricsHandler).Record(1)
startTime := time.Now().UTC()
defer func() {
metrics.AcquireShardsLatency.With(c.taggedMetricsHandler).Record(time.Since(startTime))
}()
// Readiness check: if we haven't marked readiness yet, then we need to set up a context to
// run the readiness check on owned shards.
var readinessCtx context.Context
var readinessCancel context.CancelFunc
if !c.initialShardsAcquired.Ready() {
readinessCtx, readinessCancel = context.WithCancel(ctx)
} else {
readinessCancel = func() {} // we need a non-nil func for Swap
}
// Cancel previous readiness check to ensure that the readiness check is always running on
// the most recent set of owned shards (e.g. after a membership change).
if prevCancel := c.shardReadinessCancel.Swap(readinessCancel); prevCancel != nil {
controller_impl.go
}
var ownedShards []int32 // only populated if we are doing a readiness check
tryAcquire := func(shardID int32) {
if err := c.ownership.verifyOwnership(shardID); err != nil {
if c.config.ShardLingerTimeLimit() > 0 {
c.CloseShardByID(shardID)
}
}
}
ownedShardsLock.Lock()
ownedShards = append(ownedShards, shardID)
ownedShardsLock.Unlock()
}
if err != nil {
metrics.GetEngineForShardErrorCounter.With(c.taggedMetricsHandler).Record(1)
c.contextTaggedLogger.Error("Unable to create history shard context", tag.Error(err), tag.OperationFailed, tag.ShardID(shardID))
// Wait up to 1s for the shard to acquire the rangeid lock.
// After 1s we will move on but the shard will continue trying in the background.
defer engineCancel()
_, _ = shard.GetEngine(engineCtx)
}
sem := semaphore.NewWeighted(concurrency)
numShards := c.config.NumberOfShards
randomStartOffset := rand.Int31n(numShards)
for index := range numShards {
shardID := (index+randomStartOffset)%numShards + 1
if err := sem.Acquire(ctx, 1); err != nil {
break
}
defer sem.Release(1)
tryAcquire(shardID)
}()
}
c.RLock()
// note that this count includes lingering shards
numOfOwnedShards := len(c.historyShards)
c.RUnlock()
metrics.NumShardsGauge.With(c.taggedMetricsHandler).Record(float64(numOfOwnedShards))
c.publishShardCountUpdate(numOfOwnedShards)
// Readiness check: We should set initialShardsAcquired when:
// 1. It's not already set.
// 2. We should own at least one shard (i.e. not before we join membership).
// 3. We have ownership of all the shards we're supposed to own.
if readinessCtx != nil {
if len(ownedShards) > 0 {
defer readinessCancel()
if c.checkShardReadiness(readinessCtx, ownedShards) {
c.initialShardsAcquired.SetIfNotReady(struct{}{}, nil)
}
}()
readinessCancel()
}
}
}
ctx context.Context,
shards []int32,
concurrency := int64(max(c.config.AcquireShardConcurrency(), 1))
sem := semaphore.NewWeighted(concurrency)
var ready atomic.Int32
for _, shardID := range shards {
if sem.Acquire(ctx, 1) != nil {
return false
}
defer sem.Release(1)
// Note that AssertOwnership uses a detached context for the actual persistence
// op so we can't cancel it. If context is canceled, the final Acquire will
// fail and we won't do anything.
if shard, err := c.GetShardByID(shardID); err != nil {
return
return
}
ready.Add(1)
}()
}
}
if ready.Load() != int32(len(shards)) {
// publishShardCountUpdate publishes the current number of shards that this controller owns to all shard count
// subscribers in a non-blocking manner.
c.RLock()
defer c.RUnlock()
for sub := range c.shardCountSubscriptions {
select {
case sub.ch <- shardCount:
}
if shardID <= 0 {
return invalidShardIdLowerBound
}
return invalidShardIdUpperBound
}
}
}
switch err.(type) {
case *persistence.ShardOwnershipLostError:
return true
return true
}
}
return tag.Int("numShards", n)
}
// 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
// GetHostInfo for testing
return testHostInfo
}
// GetClusterMetadata for testing
return t.ClusterMetadata
}
// GetClusterMetadata for testing
// GetNamespaceRegistry for testing
return t.NamespaceCache
}
// GetTimeSource for testing
return t.TimeSource
}
// 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
// GetHistoryServiceResolver for testing
return t.HistoryServiceResolver
}
// GetWorkerServiceResolver 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
return t.ExecutionMgr
}
// loggers
// 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
}
// 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.DebugLevel) {
msg = setDefaultMsg(msg)
fields := l.buildFieldsWithCallAt(tags)
}
if l.zl.Core().Enabled(zap.InfoLevel) {
msg = setDefaultMsg(msg)
fields := l.buildFieldsWithCallAt(tags)
l.zl.Info(msg, fields...)
}
}
//
// by deduping "foo" against any existing "foo" tags *only in the former*
cloneTags := mergeTags(l.tags, tags)
if l.baseZl == nil {
l.baseZl = l.zl
}
}
fields := make([]zap.Field, len(tags))
l.fillFields(tags, fields)
zl := l.baseZl.With(fields...)
return &zapLogger{
zl: zl,
skip: l.skip,
baseZl: l.baseZl,
tags: tags,
}
}
func (l *zapLogger) Skip(extraSkip int) Logger {
}
// Even if oldTags empty, we don't just return newTags because we need to de-dupe it.
outTags = slices.Clone(oldTags)
for _, t := range newTags {
if i := slices.IndexFunc(outTags, func(ti tag.Tag) bool {
}); i >= 0 {
outTags = append(outTags, t)
}
}
}
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
}
}
)
func NewHandler(logger log.Logger, clientConfig metrics.ClientConfig) (*Handler, error) {
metricstest.go
registry := prometheus.NewRegistry()
exporter, err := exporters.New(exporters.WithRegisterer(registry))
if err != nil {
return nil, err
}
// Set any custom histogram bucket configuration.
for _, u := range []string{metrics.Dimensionless, metrics.Bytes, metrics.Milliseconds} {
views = append(views, sdkmetrics.NewView(
sdkmetrics.Instrument{
Kind: sdkmetrics.InstrumentKindHistogram,
Unit: u,
},
sdkmetrics.Stream{
Aggregation: sdkmetrics.AggregationExplicitBucketHistogram{
Boundaries: clientConfig.PerUnitHistogramBoundaries[u],
},
},
))
}
provider := sdkmetrics.NewMeterProvider(
sdkmetrics.WithReader(exporter),
sdkmetrics.WithView(views...),
)
meter := provider.Meter("temporal")
otelHandler, err := metrics.NewOtelMetricsHandler(logger, &otelProvider{meter: meter}, clientConfig, false)
if err != nil {
return nil, err
}
Handler: otelHandler,
reg: registry,
}
return metricsHandler, nil
}
func (*Handler) Stop(log.Logger) {}
rec := httptest.NewRecorder()
req := httptest.NewRequest("GET", "/metrics", nil)
handler := http.NewServeMux()
handler.HandleFunc("/metrics", promhttp.HandlerFor(h.reg, promhttp.HandlerOpts{Registry: h.reg}).ServeHTTP)
handler.ServeHTTP(rec, req)
var tp expfmt.TextParser
families, err := tp.TextToMetricFamilies(rec.Body)
if err != nil {
return Snapshot{}, err
}
histogramSamples := map[string]histogramSample{}
for name, family := range families {
for _, m := range family.GetMetric() {
collectSamples(name, family, m, samples, histogramSamples)
}
}
samples: samples,
histogramSamples: histogramSamples,
}, nil
}
func collectSamples(name string, family *dto.MetricFamily, m *dto.Metric, samples map[string]sample, histogramSamples map[string]histogramSample) {
metricstest.go
labelvalues := map[string]string{}
for _, lp := range m.GetLabel() {
labelvalues[lp.GetName()] = lp.GetValue()
}
// This only records the last sample if there
// are multiple samples recorded.
default:
// Not yet supporting summary, untyped.
buckets := m.Histogram.GetBucket()
hbs := []HistogramBucket{}
for _, bucket := range buckets {
hb := HistogramBucket{
value: float64(bucket.GetCumulativeCount()),
upperBound: bucket.GetUpperBound(),
}
hbs = append(hbs, hb)
}
histogramSamples[name] = histogramSample{
metricType: family.GetType(),
labelValues: labelvalues,
buckets: hbs,
}
samples[name] = sample{
metricType: family.GetType(),
labelValues: labelvalues,
sampleValue: m.Counter.GetValue(),
}
case dto.MetricType_GAUGE:
samples[name] = sample{
metricType: family.GetType(),
labelValues: labelvalues,
sampleValue: m.Gauge.GetValue(),
}
}
}
}
return m.meter
}
func (m *otelProvider) Stop(log.Logger) {}
func (s Snapshot) getValue(name string, metricType dto.MetricType, tags ...metrics.Tag) (float64, error) {
metricstest.go
labelValues := map[string]string{}
for _, tag := range tags {
labelValues[tag.Key] = tag.Value
}
sample, ok := s.samples[name]
if !ok {
return 0, fmt.Errorf("%w: %q", ErrMetricNotFound, name)
}
return 0, fmt.Errorf("%w: %q is a %s, not a %s", ErrMetricTypeMismatch, name, sample.metricType, metricType)
}
return 0, fmt.Errorf("%w: %q has %v, asked for %v", ErrMetricLabelMismatch, name, sample.labelValues, labelValues)
}
}
return s.getValue(name, dto.MetricType_COUNTER, tags...)
}
func (s Snapshot) Gauge(name string, tags ...metrics.Tag) (float64, error) {
cfg ClientConfig,
shouldRecordTimerInSeconds bool,
c, err := globalRegistry.buildCatalog()
if err != nil {
return nil, fmt.Errorf("failed to build metrics catalog: %w", err)
}
l: l,
set: makeInitialSet(cfg.Tags),
provider: o,
excludeTags: configExcludeTags(cfg),
catalog: c,
gauges: new(sync.Map),
recordTimerInSeconds: shouldRecordTimerInSeconds,
}, nil
}
// WithTags creates a new Handler with the provided Tag list.
// Tags are merged with the existing tags.
newHandler := *omp
newHandler.set = newHandler.makeSet(tags)
return &newHandler
}
// Counter obtains a counter for the given name.
opts := addOptions(omp, counterOptions{}, counter)
c, err := omp.provider.GetMeter().Int64Counter(counter, opts...)
if err != nil {
omp.l.Error("error getting metric", tag.String("MetricName", counter), tag.Error(err))
return CounterFunc(func(i int64, t ...Tag) {})
}
option := metric.WithAttributeSet(omp.makeSet(t))
c.Add(context.Background(), i, option)
})
}
func (omp *otelMetricsHandler) getGaugeAdapter(gauge string) (*gaugeAdapter, error) {
otel_metrics_handler.go
if v, ok := omp.gauges.Load(gauge); ok {
}
values: make(map[attribute.Distinct]gaugeValue),
}
if v, wasLoaded := omp.gauges.LoadOrStore(gauge, adapter); wasLoaded {
return v.(*gaugeAdapter), nil
}
metric.WithFloat64Callback(adapter.callback),
}, gauge)
// Register the gauge with otel. It will call our callback when it wants to read the values.
_, err := omp.provider.GetMeter().Float64ObservableGauge(gauge, opts...)
if err != nil {
omp.gauges.Delete(gauge)
omp.l.Error("error getting metric", tag.String("MetricName", gauge), tag.Error(err))
}
}
// Gauge obtains a gauge for the given name.
adapter, err := omp.getGaugeAdapter(gauge)
if err != nil {
return GaugeFunc(func(i float64, t ...Tag) {})
}
omp: omp,
adapter: adapter,
}
}
func (a *gaugeAdapter) callback(ctx context.Context, o metric.Float64Observer) error {
otel_metrics_handler.go
a.lock.Lock()
defer a.lock.Unlock()
for _, v := range a.values {
o.Observe(v.value, metric.WithAttributeSet(v.set))
}
return nil
}
set := g.omp.makeSet(tags)
g.adapter.lock.Lock()
defer g.adapter.lock.Unlock()
g.adapter.values[set.Equivalent()] = gaugeValue{value: v, set: set}
}
// Timer obtains a timer for the given name.
if omp.recordTimerInSeconds {
return omp.timerInSeconds(timer)
}
}
func (omp *otelMetricsHandler) timerInMilliseconds(timer string) TimerIface {
otel_metrics_handler.go
opts := addOptions(omp, int64HistogramOptions{metric.WithUnit(Milliseconds)}, timer)
c, err := omp.provider.GetMeter().Int64Histogram(timer, opts...)
if err != nil {
omp.l.Error("error getting metric", tag.String("MetricName", timer), tag.Error(err))
return TimerFunc(func(i time.Duration, t ...Tag) {})
}
option := metric.WithAttributeSet(omp.makeSet(t))
c.Record(context.Background(), i.Milliseconds(), option)
})
}
// makeSet returns an otel attribute.Set with the given tags merged with the
// otelMetricsHandler's tags.
if len(tags) == 0 {
}
for i := omp.set.Iter(); i.Next(); {
attrs = append(attrs, i.Attribute())
}
attrs = append(attrs, omp.convertTag(t))
}
return attribute.NewSet(attrs...)
}
func (omp *otelMetricsHandler) convertTag(tag Tag) attribute.KeyValue {
otel_metrics_handler.go
if vals, ok := omp.excludeTags[tag.Key]; ok {
if _, ok := vals[tag.Value]; !ok {
return attribute.String(tag.Key, tagExcludedValue)
}
}
}
if len(tags) == 0 {
return *attribute.EmptySet()
}
var attrs []attribute.KeyValue
for k, v := range tags {
}
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
}
}
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
}
}
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 {
c.mu.RUnlock()
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,
}
})
}
// 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 {
}
if len(t1) == 0 {
return nil
}
for i := range t1 {
nt, _ := normalizeTag(t1[i], e)
m[nt.Key] = nt.Value
}
return m
}
// 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.
return m.recorder
}
// AssertShardOwnership mocks base method.
func (m *MockShardManager) AssertShardOwnership(ctx context.Context, request *AssertShardOwnershipRequest) error {
data_interfaces_mock.go
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "AssertShardOwnership", ctx, request)
ret0, _ := ret[0].(error)
return ret0
}
// AssertShardOwnership indicates an expected call of AssertShardOwnership.
func (mr *MockShardManagerMockRecorder) AssertShardOwnership(ctx, request any) *gomock.Call {
data_interfaces_mock.go
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AssertShardOwnership", reflect.TypeOf((*MockShardManager)(nil).AssertShardOwnership), ctx, request)
}
// Close mocks base method.
// GetOrCreateShard mocks base method.
func (m *MockShardManager) GetOrCreateShard(ctx context.Context, request *GetOrCreateShardRequest) (*GetOrCreateShardResponse, error) {
data_interfaces_mock.go
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "GetOrCreateShard", ctx, request)
ret0, _ := ret[0].(*GetOrCreateShardResponse)
ret1, _ := ret[1].(error)
return ret0, ret1
}
// GetOrCreateShard indicates an expected call of GetOrCreateShard.
func (mr *MockShardManagerMockRecorder) GetOrCreateShard(ctx, request any) *gomock.Call {
data_interfaces_mock.go
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetOrCreateShard", reflect.TypeOf((*MockShardManager)(nil).GetOrCreateShard), ctx, request)
}
// UpdateShard mocks base method.
func (m *MockShardManager) UpdateShard(ctx context.Context, request *UpdateShardRequest) error {
data_interfaces_mock.go
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "UpdateShard", ctx, request)
ret0, _ := ret[0].(error)
return ret0
}
// UpdateShard indicates an expected call of UpdateShard.
func (mr *MockShardManagerMockRecorder) UpdateShard(ctx, request any) *gomock.Call {
data_interfaces_mock.go
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UpdateShard", reflect.TypeOf((*MockShardManager)(nil).UpdateShard), ctx, request)
}
// MockExecutionManager is a mock of ExecutionManager interface.
// 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 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.
// 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.
}
if x != nil {
return x.RangeId
}
return 0
}
}
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
}
func (*TaskKey) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_queues_proto_msgTypes[0]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
}
*x = QueueState{}
mi := &file_temporal_server_api_persistence_v1_queues_proto_msgTypes[1]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *QueueState) String() string {
func (*QueueState) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_queues_proto_msgTypes[1]
if x != nil {
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
func (*QueueReaderState) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_queues_proto_msgTypes[2]
if x != nil {
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
}
func (*QueueSliceScope) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_queues_proto_msgTypes[3]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
func (*QueueSliceRange) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_queues_proto_msgTypes[4]
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
}
// 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.
return m.recorder
}
// AddListener mocks base method.
// Lookup mocks base method.
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "Lookup", key)
ret0, _ := ret[0].(HostInfo)
ret1, _ := ret[1].(error)
return ret0, ret1
}
// Lookup indicates an expected call of Lookup.
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Lookup", reflect.TypeOf((*MockServiceResolver)(nil).Lookup), key)
}
// LookupN mocks base method.
// 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 (m *MockHostInfoProvider) EXPECT() *MockHostInfoProviderMockRecorder {
interfaces_mock.go
return m.recorder
}
// HostInfo mocks base method.
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "HostInfo")
ret0, _ := ret[0].(HostInfo)
return ret0
}
// HostInfo indicates an expected call of HostInfo.
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "HostInfo", reflect.TypeOf((*MockHostInfoProvider)(nil).HostInfo))
}
// 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,
}
}
// are available or ctx is done. On success, returns nil. On failure, returns
// ctx.Err() and leaves the semaphore unchanged.
func (s *PrioritySemaphoreImpl) Acquire(ctx context.Context, priority Priority, n int) error {
priority_semaphore_impl.go
if priority >= NumPriorities {
// nolint:forbidigo
panic(fmt.Sprintf("semaphore: invalid priority %v, priority must be less than %v", priority, NumPriorities))
}
s.mu.Lock()
select {
// ctx becoming done has "happened before" acquiring the semaphore,
// whether it became done before the call began or while we were
// waiting for the mutex. We prefer to fail even if we could acquire
// the mutex without blocking.
s.mu.Unlock()
return ctx.Err()
}
// Check if acquisition can proceed without waiting
// Since we hold s.mu and haven't synchronized since checking done, if
priority_semaphore_impl.go
// ctx becomes done before we return here, it becoming done must have
// "happened concurrently" with this call - it cannot "happen before"
// we return in this branch. So, we're ok to always acquire here.
s.cur += n
s.mu.Unlock()
return nil
}
if n > s.size {
}
s.mu.Lock()
defer s.mu.Unlock()
s.cur -= n
if s.cur < 0 {
s.mu.Unlock()
panic("semaphore: released more than held")
}
}
for _, l := range s.waitLists {
for {
next := l.Front()
if next == nil {
break // No more waiters blocked.
}
// noWaiters returns if there is no waiter that has priority higher or equal to lowestPriority.
func (s *PrioritySemaphoreImpl) noWaiters(lowestPriority Priority) bool {
priority_semaphore_impl.go
for _, l := range s.waitLists[:lowestPriority+1] {
if l.Len() > 0 {
return false
}
}
}
// 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
}
}
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(
}
m.tracker.drain()
}
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(
taskMinScheduledTime time.Time,
m.generator.setTaskMinScheduledTime(taskMinScheduledTime)
}
func (m *taskKeyManager) getExclusiveReaderHighWatermark(
category tasks.Category,
minTaskKey, ok := m.tracker.minTaskKey(category)
if !ok {
}
// TODO: should this be moved generator.setTaskKeys() ?
// TODO: Truncation here is just to make sure task scheduled time has the same precision as the old logic.
// Remove this truncation once we validate the rest of the code can worker correctly with higher precision.
m.timeSource.Now().Add(m.config.TimerProcessorMaxTimeShift()).Truncate(common.ScheduledTaskMinPrecision),
)
nextTaskKey := m.generator.peekTaskKey(category)
exclusiveReaderHighWatermark := tasks.MinKey(
minTaskKey,
nextTaskKey,
)
if category.Type() == tasks.CategoryTypeScheduled {
exclusiveReaderHighWatermark.TaskID = 0
}
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
}
// 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
// Address return tag for Address
return NewStringTag("address", ad)
}
// HostID return tag for HostID
// Number returns tag for Number
return NewInt64("number", n)
}
// NextNumber returns tag for NextNumber
return NewInt64("next-number", n)
}
// ServerName returns tag for ServerName
// ShardID returns tag for ShardID
return NewInt32("shard-id", shardID)
}
// ShardTime returns tag for ShardTime
// PreviousShardRangeID returns tag for PreviousShardRangeID
return NewInt64("previous-shard-range-id", id)
}
// ShardRangeID returns tag for ShardRangeID
return NewInt64("shard-range-id", id)
}
// ShardContextState returns tag for ShardContextState
)
// 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
}
// 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
return &retrierImpl{
policy: policy,
timeSource: timeSource,
startTime: timeSource.Now(),
currentAttempt: 1,
}
}
// WithInitialInterval sets the initial interval used by ExponentialRetryPolicy for the very first retry
// 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 {
}
return t.field
}
return t.field.Key
}
func (t ZapTag) Value() any {
}
return ZapTag{
field: zap.String(key, value),
}
}
func NewStringsTag(key string, value []string) ZapTag {
}
return ZapTag{
field: zap.Int64(key, value),
}
}
return ZapTag{
field: zap.Int(key, value),
}
}
return ZapTag{
field: zap.Int32(key, value),
}
}
func NewUInt32(key string, value uint32) ZapTag {
}
return ZapTag{
field: zap.Bool(key, value),
}
}
func NewErrorTag(key string, value error) ZapTag {
}
return ZapTag{
field: zap.Duration(key, value),
}
}
func NewDurationPtrTag(key string, value *durationpb.Duration) ZapTag {
}
return NewInt(key, value)
}
func Int32(key string, value int32) ZapTag {
}
return NewDurationTag(key, value)
}
func DurationPtr(key string, value *durationpb.Duration) ZapTag {
)
return &contextFactoryImpl{
ContextFactoryParams: ¶ms,
}
}
func (c *contextFactoryImpl) CreateContext(
shardID int32,
closeCallback CloseCallback,
shard, err := newContext(
shardID,
c.EngineFactory,
c.Config,
c.PersistenceConfig,
closeCallback,
c.Logger,
c.ThrottledLogger,
c.PersistenceExecutionManager,
c.PersistenceShardManager,
c.ClientBean,
c.HistoryClient,
c.MetricsHandler,
c.EventLogger,
c.PayloadSerializer,
c.TimeSource,
c.NamespaceRegistry,
c.SaProvider,
c.SaMapperProvider,
c.ClusterMetadata,
c.ArchivalMetadata,
c.HostInfoProvider,
c.TaskCategoryRegistry,
c.EventsCache,
c.StateMachineRegistry,
c.ChasmRegistry,
c.ChasmWorkflowRegistry,
c.EndpointRegistry,
c.HandoverTrackerFactory,
)
if err != nil {
return nil, err
}
return shard, nil
}
}
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
}
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 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,
}
}
func (t *taskRequestTracker) minTaskKey(
category tasks.Category,
t.Lock()
defer t.Unlock()
pendingTasksForCategory := t.pendingTaskKeys[category]
if len(pendingTasksForCategory) == 0 {
}
minKey := tasks.MaximumKey
// otherwise inflight request can fails as those requests are conditioned on
// the current rangeID
t.Lock()
if t.inflightRequestCount == 0 {
t.Unlock()
return
}
waitCh := make(chan struct{})
}
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)
}
}
}
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
}
// 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)
}
}
}
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
}
)
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 <-ctx.Done():
var value T
value T,
err error,
// cannot directly set status to `ready`, to prevent data race in case multiple `Get` occurs
// instead set status to `setting` to prevent concurrent completion of this future
if !atomic.CompareAndSwapInt32(
&f.status,
pending,
setting,
) {
panic("future has already been completed")
}
f.err = err
atomic.CompareAndSwapInt32(&f.status, setting, ready)
close(f.readyCh)
}
}
return atomic.LoadInt32(&f.status) == ready
}
// NewMockEngine creates a new mock instance.
mock := &MockEngine{ctrl: ctrl}
mock.recorder = &MockEngineMockRecorder{mock}
return mock
}
// EXPECT returns an object that allows the caller to indicate expected use.
return m.recorder
}
// AddTasks mocks base method.
// NotifyNewTasks mocks base method.
m.ctrl.T.Helper()
m.ctrl.Call(m, "NotifyNewTasks", arg0)
}
// NotifyNewTasks indicates an expected call of NotifyNewTasks.
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "NotifyNewTasks", reflect.TypeOf((*MockEngine)(nil).NotifyNewTasks), arg0)
}
// PauseActivity mocks base method.
// Start mocks base method.
m.ctrl.T.Helper()
m.ctrl.Call(m, "Start")
}
// Start indicates an expected call of Start.
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Start", reflect.TypeOf((*MockEngine)(nil).Start))
}
// StartWorkflowExecution mocks base method.
// Stop mocks base method.
m.ctrl.T.Helper()
m.ctrl.Call(m, "Stop")
}
// Stop indicates an expected call of Stop.
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Stop", reflect.TypeOf((*MockEngine)(nil).Stop))
}
// SubscribeReplicationNotification mocks base method.
}
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
}
// 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
}
// 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)
}
// 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.
// GetAllClusterInfo mocks base method.
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "GetAllClusterInfo")
ret0, _ := ret[0].(map[string]ClusterInformation)
return ret0
}
// GetAllClusterInfo indicates an expected call of GetAllClusterInfo.
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetAllClusterInfo", reflect.TypeOf((*MockMetadata)(nil).GetAllClusterInfo))
}
// GetClusterID mocks base method.
// GetCurrentClusterName mocks base method.
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "GetCurrentClusterName")
ret0, _ := ret[0].(string)
return ret0
}
// 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.
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(
}
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(
taskMinScheduledTime time.Time,
a.taskMinScheduledTime = util.MaxTime(a.taskMinScheduledTime, taskMinScheduledTime)
}
func (a *taskKeyGenerator) generateTaskID() (int64, error) {
}
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 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
}
// 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 data == nil {
return NewDeserializationError(enumspb.ENCODING_TYPE_UNSPECIFIED, errors.New("cannot decode nil"))
}
case enumspb.ENCODING_TYPE_JSON:
return codec.NewJSONPBEncoder().Decode(data.Data, result)
err := proto.Unmarshal(data.Data, result)
if err != nil {
return NewDeserializationError(enumspb.ENCODING_TYPE_PROTO3, err)
}
default:
return NewUnknownEncodingTypeError(data.EncodingType.String(), enumspb.ENCODING_TYPE_JSON, enumspb.ENCODING_TYPE_PROTO3)
}
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_task_proto_enumTypes[1].Descriptor()
}
func (TaskType) 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
}
)
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
}
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 {
}
}
}
return string(id)
}
func (id ID) IsEmpty() bool {
}
return string(n)
}
func (n Name) IsEmpty() bool {
}
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_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
}
}
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
}
logger log.Logger,
metricsHandler metrics.Handler,
hostIdentity := hostInfoProvider.HostInfo().Identity()
logger = log.With(logger, tag.ComponentShardController, tag.Address(hostIdentity))
return &ownership{
acquireCh: make(chan struct{}, 1),
config: config,
historyServiceResolver: historyServiceResolver,
hostInfoProvider: hostInfoProvider,
logger: logger,
membershipUpdateCh: make(chan *membership.ChangedEvent, 1),
metricsHandler: metricsHandler,
}
}
func (o *ownership) start(controller *ControllerImpl) {
// controller. If membership lists another host as the owner, it returns a
// ShardOwnershipLost error with the correct owner.
ownerInfo, err := o.historyServiceResolver.Lookup(convert.Int32ToString(shardID))
if err != nil {
return err
}
if ownerInfo.Identity() != hostInfo.Identity() {
return serviceerrors.NewShardOwnershipLost(ownerInfo.Identity(), hostInfo.GetAddress())
ownership.go
}
}
}
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_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
}
}
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_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
}
}
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_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
}
}
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_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
}
}
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
}
}
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_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
}
}
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_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
}
}
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 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_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
}
// 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.
}
}
)
return Key{
FireTime: DefaultFireTime,
TaskID: taskID,
}
}
return Key{
FireTime: fireTime,
TaskID: taskID,
}
}
func ValidateKey(key Key) error {
logger log.Logger,
metricsHandler metrics.Handler,
return &Finalizer{
logger: logger,
metricsHandler: metricsHandler,
callbacks: make(map[string]func(context.Context) error),
}
}
// Register adds a callback to the finalizer.
func (f *Finalizer) Run(
timeout time.Duration,
if timeout == 0 {
f.logger.Debug("finalizer skipped: zero timeout")
return 0
}
if f.finalized {
f.logger.Warn("finalizer skipped: called more than once")
f.mu.Unlock()
return 0
}
f.mu.Unlock() // unlocking immediately to unblock any calls to Register/Deregister
totalCount := len(f.callbacks)
if totalCount == 0 {
return 0
}
f.logger.Debug("finalizer starting",
// GetValues returns header values for passed header names.
// It always returns slice of the same size as number of passed header names.
headerValues := make([]string, len(headerNames))
for i, headerName := range headerNames {
if values := metadata.ValueFromIncomingContext(ctx, headerName); len(values) > 0 {
}
}
}
// setIncomingMD sets the key-value pairs in the incoming metadata.
// Empty values are ignored.
mdIncoming, ok := metadata.FromIncomingContext(ctx)
if !ok {
}
if v != "" {
mdIncoming.Set(k, v)
}
}
}
// 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
// times. This method is thread-safe.
c.Lock()
defer c.Unlock()
r := make(catalog, len(c.definitions))
for _, d := range c.definitions {
if original, ok := r[d.name]; ok {
return nil, fmt.Errorf(
"%w: metric %q already defined with %+v. Cannot redefine with %+v",
}
}
}
def, ok := c[name]
return def, ok
}
// NewMockEngineFactory creates a new mock instance.
mock := &MockEngineFactory{ctrl: ctrl}
mock.recorder = &MockEngineFactoryMockRecorder{mock}
return mock
}
// EXPECT returns an object that allows the caller to indicate expected use.
return m.recorder
}
// CreateEngine mocks base method.
func (m *MockEngineFactory) CreateEngine(context interfaces.ShardContext) interfaces.Engine {
engine_factory_mock.go
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "CreateEngine", context)
ret0, _ := ret[0].(interfaces.Engine)
return ret0
}
// CreateEngine indicates an expected call of CreateEngine.
func (mr *MockEngineFactoryMockRecorder) CreateEngine(context any) *gomock.Call {
engine_factory_mock.go
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CreateEngine", reflect.TypeOf((*MockEngineFactory)(nil).CreateEngine), context)
}
// ThrottleRetry is a resource aware version of Retry.
// Resource exhausted error will be retried using a different throttle retry policy, instead of the specified one.
func ThrottleRetry(operation Operation, policy RetryPolicy, isRetryable IsRetryable) error {
retry.go
ctxOp := func(context.Context) error { return operation() }
return ThrottleRetryContext(context.Background(), ctxOp, policy, isRetryable)
}
policy RetryPolicy,
isRetryable IsRetryable,
var err error
var next time.Duration
if isRetryable == nil {
isRetryable = func(error) bool { return true }
}
timeSrc := clock.NewRealTimeSource()
r := NewRetrier(policy, timeSrc)
t := NewRetrier(throttleRetryPolicy, timeSrc)
for ctx.Err() == nil {
}
if next = r.NextBackOff(err); next == done {
logger log.Logger,
disabled bool,
return newEventsCache(executionManager, handler, logger, config.EventsShardLevelCacheMaxSizeBytes(), config.EventsCacheTTL(), disabled)
}
func newEventsCache(
ttl time.Duration,
disabled bool,
opts := &cache.Options{}
opts.TTL = ttl
taggedMetricHandler := metricsHandler.WithTags(metrics.CacheTypeTag(metrics.EventsCacheTypeTagValue))
return &CacheImpl{
Cache: cache.NewWithMetrics(maxSize, opts, taggedMetricHandler),
executionManager: executionManager,
metricsHandler: taggedMetricHandler,
logger: logger,
disabled: disabled,
}
}
func (e *CacheImpl) validateKey(key EventKey) bool {
ctx context.Context,
info CallerInfo,
return setIncomingMD(ctx, map[string]string{
CallerNameHeaderName: info.CallerName,
CallerTypeHeaderName: info.CallerType,
CallOriginHeaderName: info.CallOrigin,
})
}
// SetCallerName set caller name in the context.
func GetCallerInfo(
ctx context.Context,
values := GetValues(ctx, CallerNameHeaderName, CallerTypeHeaderName, CallOriginHeaderName)
return CallerInfo{
CallerName: values[0],
CallerType: values[1],
CallOrigin: values[2],
}
}
}
return Tag{Key: serviceName, Value: string(value)}
}
func ActionType(value string) Tag {
}
return Tag{Key: OperationTagName, Value: value}
}
return Tag{Key: key, Value: value}
}
return Tag{Key: CacheTypeTagName, Value: value}
}
return Tag{Key: PriorityTagName, Value: strconv.Itoa(int(value))}
}
// ReasonString is just a string but the special type is defined here to remind callers of ReasonTag to limit the
}
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,
}
}
}
}
r.isGlobalNamespace = isGlobal
}
func (r *defaultReplicationResolver) SetActiveCluster(clusterName string) {
}
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
// 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
// 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)
}
)
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]
}
d := metricDefinition{
name: name,
description: "",
unit: "",
}
for _, opt := range opts {
opt.apply(&d)
}
return d
}
return md.name
}
func (md metricDefinition) Unit() MetricUnit {
// 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,
}
}
}
}
func (t *defaultHandoverTracker) ResolvePendingTaskIDs(maxReplicationTaskID int64) {
handover_tracker.go
for _, handoverInfo := range t.handoverNamespaces {
if handoverInfo.MaxReplicationTaskID == PendingMaxReplicationTaskID {
handoverInfo.MaxReplicationTaskID = maxReplicationTaskID
}
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")
}
}
}()
category Category,
visibilityTimestamp time.Time,
return &FakeTask{
WorkflowKey: workflowKey,
TaskID: common.EmptyEventTaskID,
Version: common.EmptyVersion,
VisibilityTimestamp: visibilityTimestamp,
Category: category,
}
}
func (f *FakeTask) GetKey() Key {
}
f.TaskID = id
}
func (f *FakeTask) GetVisibilityTime() time.Time {
// 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 {
)
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
}
)
return &serializerImpl{encodingType: encodingTypeFromEnv()}
}
func (t *serializerImpl) EncodingType() enumspb.EncodingType {
}
func (t *serializerImpl) QueueStateToBlob(info *persistencespb.QueueState) (*commonpb.DataBlob, error) {
serializer.go
return encodeBlob(info, t.encodingType)
}
func (t *serializerImpl) QueueStateFromBlob(data *commonpb.DataBlob) (*persistencespb.QueueState, error) {
serializer.go
result := &persistencespb.QueueState{}
return result, Decode(data, result)
}
// ReencodeEventBlobsAsProto3 re-encodes event blobs as proto3 if the serializer uses a different encoding.
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 {
)
if value := c.Context.Value(key); value != nil {
}
return c.valueCtx.Value(key)
// CopyContextValues copies values in source Context to destination Context.
return &valueCopyCtx{
Context: dst,
valueCtx: src,
}
}
// ResetContextTimeout creates new context with specified timeout and copies values from source Context.
// 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.
// 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())
}
// 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.
// GetAddress returns the value of the hostAddress.
return string(a)
}
// Identity returns the value of the hostAddress.
return string(a)
}
)
// 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.
var NoopCounterMetricFunc = CounterFunc(func(i int64, t ...Tag) {})
var NoopHistogramMetricFunc = HistogramFunc(func(i int64, t ...Tag) {})
)
out := make([]string, len(fields))
for i, field := range fields {
out[i] = prefix + field
}
return out
}
return strings.Join(appendPrefix(":", fields), ", ")
}
}
return durationpb.New(td)
}
func MinDurationPtr(d1 *durationpb.Duration, d2 *durationpb.Duration) *durationpb.Duration {
}
return durationMultipleOf(int64(d), time.Hour*24)
}
return DurationPtr(time.Duration(amt) * mult)
}
// ValidateAndCapProtoDuration validates protobuf durations for two conditions:
}
return TimePtr(time.Now().UTC())
}
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) {
// 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
}
// 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.
workflowID string,
runID string,
return WorkflowKey{
NamespaceID: namespaceID,
WorkflowID: workflowID,
RunID: runID,
}
}
func (k *WorkflowKey) GetNamespaceID() string {
)
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) {
// NewShardOwnershipLost returns new ShardOwnershipLost error.
func NewShardOwnershipLost(ownerHost string, currentHost string) error {
shard_ownership_lost.go
return &ShardOwnershipLost{
Message: fmt.Sprintf("Shard is owned by:%v but not by %v", ownerHost, currentHost),
OwnerHost: ownerHost,
CurrentHost: currentHost,
}
}
// Error returns string message.
// 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 strconv.FormatInt(v, 10)
}
return Int64ToString(int64(v))
}
func Uint16ToString(v uint16) string {
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)
}
)
func addOptions[T optionSet[T]](omp *otelMetricsHandler, opts T, metricName string) T {
otel_options.go
metricDef, ok := omp.catalog.getMetric(metricName)
if !ok {
return opts
}
opts = opts.addOption(metric.WithDescription(description))
}
opts = opts.addOption(metric.WithUnit(string(unit)))
}
}
// 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
}
sql.RegisterPlugin(PluginName, &plugin{
queryConverter: &queryConverter{},
connPool: newConnPool(),
})
}
func (p *plugin) GetVisibilityQueryConverter() sqlplugin.VisibilityQueryConverter {
func ConvertWeightsToDynamicConfigValue(
weights map[tasks.Priority]int,
weightsForDC := make(map[string]any)
for priority, weight := range weights {
weightsForDC[priority.String()] = weight
}
return weightsForDC
}
// 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{
// 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.
// 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.
)
func NewHistoryBranchUtil(serializer serialization.Serializer) *HistoryBranchUtilImpl {
history_branch_util.go
return &HistoryBranchUtilImpl{
serializer: serializer,
}
}
func (u *HistoryBranchUtilImpl) NewHistoryBranch(
// 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.
// With returns Logger instance that prepend every log entry with tags. If logger implements
// WithLogger it is used, otherwise every log call will be intercepted.
if wl, ok := logger.(WithLogger); ok {
}
return newWithLogger(logger, tags...)
}
}
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
// 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) {}
)
func (c HistogramFunc) Record(v int64, tags ...Tag) { c(v, tags...) }
}
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")
}
}
func ReplicationReaderIDToClusterShardID(
readerID int64,
return readerID >> 32, int32(readerID & 0xffffffff)
}
func getMinTaskKey(
}
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() {}