Atlas › Test
Exact test identity: go.temporal.io/server/common/persistence/tests/TestPGXSP/TestPostgreSQLHistoryStoreSuite/TestAppendForkSelectTrim_NonLastBranch
go.temporal.io/server/common/persistence/testsTestPGXSP/TestPostgreSQLHistoryStoreSuite/TestAppendForkSelectTrim_NonLastBranchTestAppendForkSelectTrim_NonLastBranchExpand a file to inspect source; the > gutter marks covered lines.
ctx context.Context,
request *ForkHistoryBranchRequest,
if request.ForkNodeID <= 1 {
return nil, &InvalidPersistenceRequestError{
Msg: "ForkNodeID must be > 1",
}
forkBranch, err := m.GetHistoryBranchUtil().ParseHistoryBranchInfo(request.ForkBranchToken)
history_manager.go
if err != nil {
return nil, serviceerror.NewInvalidArgument(fmt.Sprintf("unable to parse branch token: %v", err))
}
newAncestors := make([]*persistencespb.HistoryBranchRange, 0, len(forkBranch.Ancestors)+1)
history_manager.go
beginNodeID := GetBeginNodeID(forkBranch)
if beginNodeID >= request.ForkNodeID {
// this is the case that new branch's ancestors doesn't include the forking branch
for _, br := range forkBranch.Ancestors {
}
}
// this is the case the new branch will inherit all ancestors from forking branch
newAncestors = forkBranch.Ancestors
newAncestors = append(newAncestors, &persistencespb.HistoryBranchRange{
BranchId: forkBranch.GetBranchId(),
BeginNodeId: beginNodeID,
EndNodeId: request.ForkNodeID,
})
}
newBranchInfo := &persistencespb.HistoryBranch{
TreeId: forkBranch.TreeId,
BranchId: uuid.NewString(),
Ancestors: newAncestors,
}
// The above newBranchInfo is a lossy construction of the forked branch token from the original opaque branch token.
// It only initializes with the fields it understands, which may inadvertently discard other misc fields. The
// following is the replacement logic to correctly apply the updated fields into the original opaque branch token.
newBranchToken, err := m.GetHistoryBranchUtil().UpdateHistoryBranchInfo(
request.ForkBranchToken,
newBranchInfo,
request.NewRunID,
)
if err != nil {
return nil, err
}
BranchToken: newBranchToken,
BranchInfo: newBranchInfo,
ForkTime: timestamp.TimeNowPtrUtc(),
Info: request.Info,
}
treeInfoBlob, err := m.serializer.HistoryTreeInfoToBlob(treeInfo)
if err != nil {
return nil, err
}
NewBranchToken: newBranchToken,
ForkBranchInfo: forkBranch,
TreeInfo: treeInfoBlob,
ForkNodeID: request.ForkNodeID,
NewBranchID: newBranchInfo.BranchId,
Info: request.Info,
ShardID: request.ShardID,
}
err = m.persistence.ForkHistoryBranch(ctx, req)
if err != nil {
return nil, err
}
NewBranchToken: newBranchToken,
}, nil
}
ctx context.Context,
request *TrimHistoryBranchRequest,
shardID := request.ShardID
minNodeID := common.FirstEventID
maxNodeID := request.NodeID + 1
pageSize := trimHistoryBranchPageSize
branch, err := m.GetHistoryBranchUtil().ParseHistoryBranchInfo(request.BranchToken)
if err != nil {
return nil, fmt.Errorf("unable to parse history branch info: %w", err)
}
branchID := branch.BranchId
branchAncestors := branch.Ancestors
// merge tree ID & branch ID into branch ancestors so the processing logic is simple
beginNodeID := common.FirstEventID
if len(branch.Ancestors) > 0 {
beginNodeID = branch.Ancestors[len(branch.Ancestors)-1].GetEndNodeId()
}
branchAncestors = append(branchAncestors, &persistencespb.HistoryBranchRange{
history_manager.go
BranchId: branchID,
BeginNodeId: beginNodeID,
EndNodeId: maxNodeID,
})
var pageToken []byte
transactionIDToNode := map[int64]historyNodeMetadata{}
for doContinue := true; doContinue; doContinue = len(pageToken) > 0 {
token, err := m.deserializeToken(pageToken, minNodeID-1, defaultLastTransactionID)
if err != nil {
return nil, fmt.Errorf("unable to deserialize token: %w", err)
}
ctx,
request.BranchToken,
shardID,
branchAncestors,
minNodeID,
maxNodeID,
token,
pageSize,
true,
)
if err != nil {
return nil, fmt.Errorf("unable to read raw history branch: %w", err)
}
for _, node := range nodes {
transactionIDToNode[node.TransactionID] = historyNodeMetadata{
branchInfo: &persistencespb.HistoryBranch{
TreeId: treeID,
BranchId: branchID,
Ancestors: branchAncestors[0:token.CurrentRangeIndex],
},
nodeID: node.NodeID,
transactionID: node.TransactionID,
prevTransactionID: node.PrevTransactionID,
}
}
if err != nil {
return nil, fmt.Errorf("unable to serialize token: %w", err)
}
}
request.NodeID,
request.TransactionID,
transactionIDToNode,
)
if err != nil {
m.logger.Debug("unable to trim history branch due to existing history node not fully onboarded", tag.Error(err))
return &TrimHistoryBranchResponse{}, nil
}
if err := m.persistence.DeleteHistoryNodes(ctx, &InternalDeleteHistoryNodesRequest{
BranchToken: request.BranchToken,
ShardID: shardID,
BranchInfo: node.branchInfo,
NodeID: node.nodeID,
TransactionID: node.transactionID,
}); err != nil {
return nil, fmt.Errorf("unable to delete history nodes: %w", err)
}
}
}
func (m *executionManagerImpl) serializeAppendHistoryNodesRequest(
request *AppendHistoryNodesRequest,
branch, err := m.GetHistoryBranchUtil().ParseHistoryBranchInfo(request.BranchToken)
if err != nil {
return nil, serviceerror.NewInvalidArgument(fmt.Sprintf("unable to parse branch token: %v", err))
}
return nil, &InvalidPersistenceRequestError{
Msg: "events to be appended cannot be empty",
}
}
version := request.Events[0].Version
nodeID := request.Events[0].EventId
lastID := nodeID - 1
if nodeID <= 0 {
return nil, &InvalidPersistenceRequestError{
Msg: "eventID cannot be less than 1",
}
}
if e.Version != version {
return nil, &InvalidPersistenceRequestError{
Msg: "event version must be the same inside a batch",
}
}
return nil, &InvalidPersistenceRequestError{
Msg: "event ID must be continous",
}
}
}
// nodeID will be the first eventID
if err != nil {
return nil, err
}
sizeLimit := m.transactionSizeLimit()
if size > sizeLimit {
return nil, &TransactionSizeLimitError{
Msg: fmt.Sprintf("transaction size of %v bytes exceeds limit of %v bytes", size, sizeLimit),
}
BranchToken: request.BranchToken,
IsNewBranch: request.IsNewBranch,
Info: request.Info,
BranchInfo: branch,
Node: InternalHistoryNode{
NodeID: nodeID,
Events: blob,
PrevTransactionID: request.PrevTransactionID,
TransactionID: request.TransactionID,
},
ShardID: request.ShardID,
}
if req.IsNewBranch {
treeInfoBlob, err := m.serializer.HistoryTreeInfoToBlob(&persistencespb.HistoryTreeInfo{
BranchToken: request.BranchToken, // NOTE: this is redundant but double-writing until 1 minor release later
BranchInfo: branch,
ForkTime: timestamp.TimeNowPtrUtc(),
Info: request.Info,
})
if err != nil {
return nil, err
}
}
return nil, &InvalidPersistenceRequestError{
Msg: "cannot append to ancestors' nodes",
ctx context.Context,
request *AppendHistoryNodesRequest,
req, err := m.serializeAppendHistoryNodesRequest(request)
if err != nil {
return nil, err
}
return &AppendHistoryNodesResponse{
Size: len(req.Node.Events.Data),
}, err
}
ctx context.Context,
request *ReadHistoryBranchRequest,
resp := &ReadHistoryBranchResponse{}
var err error
resp.HistoryEvents, _, _, resp.NextPageToken, resp.Size, err = m.readHistoryBranch(ctx, false, request)
return resp, err
}
// ReadRawHistoryBranch returns raw history binary data for a branch
pageSize int,
metadataOnly bool,
if token.CurrentRangeIndex == notStartedIndex {
for idx, br := range branchAncestors {
// this range won't contain any nodes needed
if minNodeID >= br.GetEndNodeId() {
continue
}
// similarly, the ranges and the rest won't contain any nodes needed,
break
}
token.CurrentRangeIndex = idx
}
token.FinalRangeIndex = idx
}
return nil, nil, softassert.UnexpectedDataLoss(m.logger, "branchRange is corrupted", nil)
}
}
// minNodeID remains the same, since caller can read from the middle
// maxNodeID need to be shortened since this branch can contain additional history nodes
if currentBranch.GetEndNodeId() < maxNodeID {
}
resp, err := m.persistence.ReadHistoryBranch(ctx, &InternalReadHistoryBranchRequest{
BranchToken: branchToken,
ShardID: shardID,
BranchID: branchID,
MinNodeID: minNodeID,
MaxNodeID: maxNodeID,
NextPageToken: token.StoreToken,
PageSize: pageSize,
MetadataOnly: metadataOnly,
})
if err != nil {
return nil, nil, err
}
return resp.Nodes, token, nil
}
ctx context.Context,
request *ReadHistoryBranchRequest,
) ([]*commonpb.DataBlob, []int64, []int64, *historyPagingToken, int, error) {
history_manager.go
shardID := request.ShardID
branchToken := request.BranchToken
minNodeID := request.MinEventID
maxNodeID := request.MaxEventID
branch, err := m.GetHistoryBranchUtil().ParseHistoryBranchInfo(branchToken)
if err != nil {
return nil, nil, nil, nil, 0, serviceerror.NewInvalidArgument(fmt.Sprintf("unable to parse branch token: %v", err))
}
branchAncestors := branch.Ancestors
// merge tree ID & branch ID into branch ancestors so the processing logic is simple
beginNodeID := common.FirstEventID
if len(branch.Ancestors) > 0 {
}
branchAncestors = append(branchAncestors, &persistencespb.HistoryBranchRange{
history_manager.go
BranchId: branchID,
BeginNodeId: beginNodeID,
EndNodeId: maxNodeID,
})
token, err := m.deserializeToken(
request.NextPageToken,
request.MinEventID-1,
defaultLastTransactionID,
)
if err != nil {
return nil, nil, nil, nil, 0, err
}
ctx,
branchToken,
shardID,
branchAncestors,
minNodeID,
maxNodeID,
token,
request.PageSize,
false,
)
if err != nil {
return nil, nil, nil, nil, 0, err
}
return nil, nil, nil, nil, 0, serviceerror.NewNotFound("Workflow execution history not found.")
}
token.LastNodeID,
token.LastTransactionID,
nodes,
)
if err != nil {
return nil, nil, nil, nil, 0, err
}
transactionIDs := make([]int64, 0, len(nodes))
nodeIDs := make([]int64, 0, len(nodes))
dataSize := 0
if len(nodes) > 0 {
dataBlobs = make([]*commonpb.DataBlob, len(nodes))
for index, node := range nodes {
dataBlobs[index] = node.Events
if node.Events == nil {
return nil, nil, nil, nil, 0, softassert.UnexpectedDataLoss(m.logger, "no events in history node", nil)
}
transactionIDs = append(transactionIDs, node.TransactionID)
nodeIDs = append(nodeIDs, node.NodeID)
}
token.LastNodeID = lastNode.NodeID
token.LastTransactionID = lastNode.TransactionID
}
}
byBatch bool,
request *ReadHistoryBranchRequest,
) ([]*historypb.HistoryEvent, []*historypb.History, []int64, []byte, int, error) {
history_manager.go
dataBlobs, transactionIDs, _, token, dataSize, err := m.readRawHistoryBranchAndFilter(ctx, request)
if err != nil {
return nil, nil, nil, nil, 0, err
}
historyEventBatches := make([]*historypb.History, 0, request.PageSize)
var firstEvent, lastEvent *historypb.HistoryEvent
var eventCount int
dataLossTags := func(cause error) []tag.Tag {
return []tag.Tag{
tag.Cause(cause.Error()),
}
events, err := m.serializer.DeserializeEvents(batch)
if err != nil {
return nil, nil, nil, nil, dataSize, err
}
return nil, nil, nil, nil, dataSize, softassert.UnexpectedDataLoss(m.logger, dataLossMsg, errEmptyEvents, dataLossTags(errEmptyEvents)...)
}
eventCount = len(events)
lastEvent = events[eventCount-1]
if firstEvent.GetVersion() != lastEvent.GetVersion() || firstEvent.GetEventId()+int64(eventCount-1) != lastEvent.GetEventId() {
// in a single batch, version should be the same, and ID should be contiguous
return historyEvents, historyEventBatches, transactionIDs, nil, dataSize, softassert.UnexpectedDataLoss(m.logger, dataLossMsg, errWrongVersion, dataLossTags(errWrongVersion)...)
}
return historyEvents, historyEventBatches, transactionIDs, nil, dataSize, softassert.UnexpectedDataLoss(m.logger, dataLossMsg, errNonContiguousEventID, dataLossTags(errNonContiguousEventID)...)
}
historyEventBatches = append(historyEventBatches, &historypb.History{Events: events})
historyEvents = append(historyEvents, events...)
}
token.LastEventID = lastEvent.GetEventId()
}
if err != nil {
return nil, nil, nil, nil, 0, err
}
return historyEvents, historyEventBatches, transactionIDs, nextPageToken, dataSize, nil
history_manager.go
}
lastTransactionID int64,
nodes []InternalHistoryNode,
var result []InternalHistoryNode
for _, node := range nodes {
// assuming that business logic layer is correct and transaction ID only increase
// thus, valid event batch will come with increasing transaction ID
// event batches with smaller node ID
// -> should not be possible since records are already sorted
// event batches with same node ID
// -> batch with higher transaction ID is valid
// event batches with larger node ID
// -> batch with lower transaction ID is invalid (happens before)
// -> batch with higher transaction ID is valid
if node.TransactionID < lastTransactionID {
continue
}
case node.NodeID < lastNodeID:
return nil, softassert.UnexpectedDataLoss(m.logger, "corrupted data, nodeID cannot decrease", nil)
case node.NodeID == lastNodeID:
return nil, softassert.UnexpectedDataLoss(m.logger, "corrupted data, same nodeID must have smaller txnID", nil)
// NOTE: when row.nodeID > lastNodeID, we expect the one with largest txnID comes first
lastTransactionID = node.TransactionID
lastNodeID = node.NodeID
result = append(result, node)
}
}
}
defaultLastEventID int64,
lastTransactionId int64,
return m.pagingTokenSerializer.Deserialize(
token,
defaultLastEventID,
defaultLastNodeID,
lastTransactionId,
)
}
func (m *executionManagerImpl) serializeToken(
pagingToken *historyPagingToken,
reverseOrder bool,
if len(pagingToken.StoreToken) == 0 {
if pagingToken.CurrentRangeIndex == pagingToken.FinalRangeIndex {
// this means that we have reached the final page of final branchRange
return nil, nil
}
pagingToken.CurrentRangeIndex--
pagingToken.CurrentRangeIndex++
}
return m.pagingTokenSerializer.Serialize(pagingToken)
}
}
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 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 IntPropertyFn = TypedPropertyFn[int]
return GetTypedPropertyFn(value)
}
type NamespaceIntSetting = NamespaceTypedSetting[int]
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 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 {
}
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 {
}
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 {
// 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 {
// 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 {
// 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 {
// 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 {
// 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 {
store p.ExecutionStore,
logger log.Logger,
serializer := serialization.NewSerializer()
return &HistoryEventsSuite{
Assertions: require.New(t),
ProtoAssertions: protorequire.New(t),
store: p.NewExecutionManager(
store,
serializer,
nil,
logger,
dynamicconfig.GetIntPropertyFn(4*1024*1024),
dynamicconfig.GetBoolPropertyFn(false),
),
serializer: serializer,
logger: logger,
}
}
}
}
s.Assertions = require.New(s.T())
s.ProtoAssertions = protorequire.New(s.T())
s.Ctx, s.Cancel = context.WithTimeout(context.Background(), 30*time.Second*debug.TimeoutMultiplier)
s.ShardID++
}
s.Cancel()
}
func (s *HistoryEventsSuite) TestAppendSelect_First() {
}
treeID := uuid.NewString()
branchID := uuid.NewString()
branchToken, err := s.store.GetHistoryBranchUtil().NewHistoryBranch(
uuid.NewString(),
uuid.NewString(),
uuid.NewString(),
treeID,
&branchID,
[]*persistencespb.HistoryBranchRange{},
time.Duration(0),
time.Duration(0),
time.Duration(0),
)
s.NoError(err)
var events0 []*historypb.HistoryEvent
var events1 []*historypb.HistoryEvent
eventsPacket0 := s.newHistoryEvents(
[]int64{1, 2, 3},
rand.Int63(),
0,
)
s.appendHistoryEvents(s.ShardID, branchToken, eventsPacket0)
events0 = append(events0, eventsPacket0.events...)
events1 = append(events1, eventsPacket0.events...)
eventsPacket1 := s.newHistoryEvents(
[]int64{4, 5},
eventsPacket0.transactionID+1,
eventsPacket0.transactionID,
)
s.appendHistoryEvents(s.ShardID, branchToken, eventsPacket1)
events0 = append(events0, eventsPacket1.events...)
events1 = append(events1, eventsPacket1.events...)
s.appendHistoryEvents(s.ShardID, branchToken, s.newHistoryEvents(
[]int64{4, 5},
eventsPacket0.transactionID+2,
eventsPacket0.transactionID,
))
eventsPacket20 := s.newHistoryEvents(
[]int64{6},
eventsPacket1.transactionID+2,
eventsPacket1.transactionID,
)
s.appendHistoryEvents(s.ShardID, branchToken, eventsPacket20)
events0 = append(events0, eventsPacket20.events...)
newBranchToken := s.forkHistoryBranch(s.ShardID, branchToken, 6)
eventsPacket21 := s.newHistoryEvents(
[]int64{6},
eventsPacket1.transactionID+3,
eventsPacket1.transactionID,
)
s.appendHistoryEvents(s.ShardID, newBranchToken, eventsPacket21)
events1 = append(events1, eventsPacket21.events...)
if rand.Intn(2)%2 == 0 {
s.trimHistoryBranch(s.ShardID, branchToken, eventsPacket20.nodeID, eventsPacket20.transactionID)
history_store.go
s.trimHistoryBranch(s.ShardID, newBranchToken, eventsPacket21.nodeID, eventsPacket21.transactionID)
}
protorequire.ProtoSliceEqual(s.T(), events0, s.listAllHistoryEvents(s.ShardID, branchToken))
history_store.go
protorequire.ProtoSliceEqual(s.T(), events1, s.listAllHistoryEvents(s.ShardID, newBranchToken))
}
branchToken []byte,
packet HistoryEventsPacket,
_, err := s.store.AppendHistoryNodes(s.Ctx, &p.AppendHistoryNodesRequest{
ShardID: shardID,
BranchToken: branchToken,
Events: packet.events,
TransactionID: packet.transactionID,
PrevTransactionID: packet.prevTransactionID,
IsNewBranch: packet.nodeID == common.FirstEventID,
Info: "",
})
s.NoError(err)
}
func (s *HistoryEventsSuite) appendRawHistoryBatches(
branchToken []byte,
newNodeID int64,
resp, err := s.store.ForkHistoryBranch(s.Ctx, &p.ForkHistoryBranchRequest{
ShardID: shardID,
NamespaceID: uuid.NewString(),
ForkBranchToken: branchToken,
ForkNodeID: newNodeID,
Info: "",
NewRunID: uuid.NewString(),
})
s.NoError(err)
return resp.NewBranchToken
}
func (s *HistoryEventsSuite) deleteHistoryBranch(
nodeID int64,
transactionID int64,
_, err := s.store.TrimHistoryBranch(s.Ctx, &p.TrimHistoryBranchRequest{
ShardID: shardID,
BranchToken: branchToken,
NodeID: nodeID,
TransactionID: transactionID,
})
s.NoError(err)
}
func (s *HistoryEventsSuite) listHistoryEvents(
shardID int32,
branchToken []byte,
var token []byte
var events []*historypb.HistoryEvent
for doContinue := true; doContinue; doContinue = len(token) > 0 {
resp, err := s.store.ReadHistoryBranch(s.Ctx, &p.ReadHistoryBranchRequest{
ShardID: shardID,
BranchToken: branchToken,
MinEventID: common.FirstEventID,
MaxEventID: common.LastEventID,
PageSize: 1, // use 1 here for better testing exp
NextPageToken: token,
})
s.NoError(err)
token = resp.NextPageToken
events = append(events, resp.HistoryEvents...)
}
return events
}
transactionID int64,
prevTransactionID int64,
events := make([]*historypb.HistoryEvent, len(eventIDs))
for index, eventID := range eventIDs {
events[index] = &historypb.HistoryEvent{
EventId: eventID,
EventTime: timestamppb.New(time.Unix(0, rand.Int63()).UTC()),
}
}
nodeID: eventIDs[0],
transactionID: transactionID,
prevTransactionID: prevTransactionID,
events: events,
}
}
ctx context.Context,
request *p.InternalAppendHistoryNodesRequest,
branchInfo := request.BranchInfo
node := request.Node
treeIDBytes, err := primitives.ParseUUID(branchInfo.GetTreeId())
if err != nil {
return err
}
if err != nil {
return err
}
TreeID: treeIDBytes,
BranchID: branchIDBytes,
NodeID: node.NodeID,
PrevTxnID: node.PrevTransactionID,
TxnID: node.TransactionID,
Data: node.Events.Data,
DataEncoding: node.Events.EncodingType.String(),
ShardID: request.ShardID,
}
if !request.IsNewBranch {
switch err {
case nil:
return nil
case context.DeadlineExceeded, context.Canceled:
return &p.AppendHistoryTimeoutError{
}
treeRow := &sqlplugin.HistoryTreeRow{
ShardID: request.ShardID,
TreeID: treeIDBytes,
BranchID: branchIDBytes,
Data: treeInfoBlob.Data,
DataEncoding: treeInfoBlob.EncodingType.String(),
}
return m.txExecute(ctx, "AppendHistoryNodes", func(tx sqlplugin.Tx) error {
result, err := tx.InsertIntoHistoryNode(ctx, nodeRow)
if err != nil {
return err
}
if err != nil {
return err
}
return fmt.Errorf("expected 1 or 2 row to be affected for node table, got %v", rowsAffected)
}
switch err {
case nil:
rowsAffected, err = result.RowsAffected()
if err != nil {
return err
}
return fmt.Errorf("expected 1 or 2 rows to be affected for tree table as we allow upserts, got %v", rowsAffected)
}
case context.DeadlineExceeded, context.Canceled:
return &p.AppendHistoryTimeoutError{
ctx context.Context,
request *p.InternalDeleteHistoryNodesRequest,
branchInfo := request.BranchInfo
nodeID := request.NodeID
txnID := request.TransactionID
shardID := request.ShardID
if nodeID < p.GetBeginNodeID(branchInfo) {
return &p.InvalidPersistenceRequestError{
Msg: "cannot append to ancestors' nodes",
}
if err != nil {
return err
}
if err != nil {
return err
}
TreeID: treeIDBytes,
BranchID: branchIDBytes,
NodeID: nodeID,
TxnID: txnID,
ShardID: shardID,
}
_, err = m.DB.DeleteFromHistoryNode(ctx, nodeRow)
if err != nil {
return serviceerror.NewUnavailablef("DeleteHistoryNodes: %v", err)
}
}
ctx context.Context,
request *p.InternalReadHistoryBranchRequest,
branch, err := m.ParseHistoryBranchInfo(request.BranchToken)
if err != nil {
return nil, err
}
if err != nil {
return nil, err
}
if err != nil {
return nil, err
}
if len(request.NextPageToken) == 0 {
if request.ReverseOrder {
token = &historyNodePaginationToken{LastNodeID: request.MaxNodeID, LastTxnID: MaxTxnID}
token = &historyNodePaginationToken{LastNodeID: request.MinNodeID, LastTxnID: MinTxnID}
}
token, err = deserializePageTokenJson[historyNodePaginationToken](request.NextPageToken)
if err != nil {
return nil, err
}
}
minTxnId, maxTxnId := MinTxnID, MaxTxnID
if request.ReverseOrder {
maxNodeId = token.LastNodeID
maxTxnId = token.LastTxnID
minNodeId = token.LastNodeID
minTxnId = token.LastTxnID
}
rows, err := m.DB.RangeSelectFromHistoryNode(ctx, sqlplugin.HistoryNodeSelectFilter{
history_store.go
ShardID: request.ShardID,
TreeID: treeIDBytes,
BranchID: branchIDBytes,
MinNodeID: minNodeId,
MinTxnID: minTxnId,
MaxNodeID: maxNodeId,
MaxTxnID: maxTxnId,
PageSize: request.PageSize,
MetadataOnly: request.MetadataOnly,
ReverseOrder: request.ReverseOrder,
})
switch err {
case nil:
// noop
case sql.ErrNoRows:
}
for _, row := range rows {
nodes = append(nodes, p.InternalHistoryNode{
NodeID: row.NodeID,
PrevTransactionID: row.PrevTxnID,
TransactionID: row.TxnID,
Events: p.NewDataBlob(row.Data, row.DataEncoding),
})
}
if len(rows) < request.PageSize {
pagingToken = nil
} else {
pagingToken, err = serializePageTokenJson(&historyNodePaginationToken{
LastNodeID: lastRow.NodeID,
LastTxnID: lastRow.TxnID,
})
if err != nil {
return nil, err
}
}
Nodes: nodes,
NextPageToken: pagingToken,
}, nil
}
ctx context.Context,
request *p.InternalForkHistoryBranchRequest,
forkB := request.ForkBranchInfo
treeInfoBlob := request.TreeInfo
newBranchIdBytes, err := primitives.ParseUUID(request.NewBranchID)
if err != nil {
return err
}
if err != nil {
return err
}
ShardID: request.ShardID,
TreeID: treeIDBytes,
BranchID: newBranchIdBytes,
Data: treeInfoBlob.Data,
DataEncoding: treeInfoBlob.EncodingType.String(),
}
result, err := m.DB.InsertIntoHistoryTree(ctx, row)
if err != nil {
return err
}
if err != nil {
return err
}
return fmt.Errorf("expected 1 or 2 row to be affected for tree table, got %v", rowsAffected)
}
}
func LoadAndSplitQuery(
filePaths []string,
var files []io.Reader
for _, filePath := range filePaths {
f, err := os.Open(filePath)
if err != nil {
return nil, fmt.Errorf("error opening file %s: %w", filePath, err)
}
}
}
func LoadAndSplitQueryFromReaders(
readers []io.Reader,
result := make([]string, 0, querySliceDefaultSize)
for _, r := range readers {
content, err := io.ReadAll(r)
if err != nil {
return nil, fmt.Errorf("error reading contents: %w", err)
}
contentStr := string(bytes.ToLower(content))
for i, j := 0, 0; i < n; i = j {
// stack to keep track of open parenthesis/blocks
var st []byte
var stmtBuilder strings.Builder
stmtLoop:
for ; j < n; j++ {
switch contentStr[j] {
if len(st) == 0 {
j++
break stmtLoop
}
st = append(st, sqlLeftParenthesis)
if len(st) == 0 || st[len(st)-1] != sqlLeftParenthesis {
return nil, fmt.Errorf("error reading contents: unmatched right parenthesis")
}
if !hasWordAt(contentStr, sqlDoubleDollarKeyword, j) {
continue
}
st = append(st, sqlDoubleDollarKeyword[0])
j += len(sqlDoubleDollarKeyword) - 1
} else {
st = st[:len(st)-1]
j += len(sqlDoubleDollarKeyword) - 1
}
if !hasWordAt(contentStr, sqlIfKeyword, j) {
continue
}
hasWordsBefore(contentStr, j-1, sqlCreateKeyword, sqlIndexKeyword) ||
hasWordsBefore(contentStr, j-1, sqlCreateKeyword, sqlIndexKeyword, sqlConcurrentlyKeyword) ||
hasWordsBefore(contentStr, j-1, sqlCreateKeyword, sqlTableKeyword) {
continue
}
j += len(sqlIfKeyword) - 1
if !hasWordAt(contentStr, sqlLoopKeyword, j) {
continue
}
st = append(st, sqlLoopKeyword[0])
j += len(sqlLoopKeyword) - 1
if hasWordAt(contentStr, sqlBeginKeyword, j) {
j += len(sqlBeginKeyword) - 1
}
if !hasWordAt(contentStr, sqlEndKeyword, j) {
continue
}
if ok, after := hasWordAfter(contentStr, sqlIfKeyword, j+len(sqlEndKeyword)); ok {
query_util.go
return nil, errors.New("error reading contents: unmatched `END IF` keyword")
}
j = after + len(sqlIfKeyword) - 1
} else if ok, after := hasWordAfter(contentStr, sqlLoopKeyword, j+len(sqlEndKeyword)); ok {
query_util.go
//nolint:revive
if len(st) == 0 || st[len(st)-1] != sqlLoopKeyword[0] {
st = st[:len(st)-1]
j = after + len(sqlLoopKeyword) - 1
if len(st) == 0 || st[len(st)-1] != sqlBeginKeyword[0] {
return nil, errors.New("error reading contents: unmatched `END` keyword")
}
j += len(sqlEndKeyword) - 1
}
quote := contentStr[j]
j++
for j < n && contentStr[j] != quote {
j++
}
if j == n {
return nil, fmt.Errorf("error reading contents: unmatched quotes")
}
if j+len(sqlLineComment) <= n && contentStr[j:j+len(sqlLineComment)] == sqlLineComment {
_, _ = stmtBuilder.Write(bytes.TrimRight(content[i:j], " "))
for j < n && contentStr[j] != '\n' {
j++
}
i = j
}
// no-op: generic character
}
}
switch st[len(st)-1] {
case sqlLeftParenthesis:
}
stmt := strings.TrimSpace(stmtBuilder.String())
if stmt == "" {
}
}
}
}
// hasWordAt is a simple test to check if it matches the whole word:
// it checks if the adjacent characters are not alphanumeric if they exist.
if pos+len(word) > len(s) || s[pos:pos+len(word)] != word {
return false
}
}
}
}
// hasWordAfter checks if the given word appears after position pos in s,
// separated by at least one space, and is a whole word.
after := pos
for after < len(s) && unicode.IsSpace(rune(s[after])) {
}
return false, after
}
}
// Words must appear is the given order.
// Eg: hasWordsBefore("CREATE TABLE IF", 12, "CREATE", "TABLE") returns true.
if len(words) == 0 {
return true
}
return false
}
// skip spaces
for pos >= 0 && unicode.IsSpace(rune(s[pos])) {
pos--
}
// go to first char of word
pos--
}
// check if current pos matches the word
return false
}
pos--
}
}
return unicode.IsLetter(rune(c)) || unicode.IsDigit(rune(c))
}
}
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.
// NewSearchAttributeDateTime creates a new date time search attribute given a predefined chasm field
func NewSearchAttributeDateTime(alias string, datetimeField SearchAttributeFieldDateTime) SearchAttributeDateTime {
search_attribute.go
return SearchAttributeDateTime{
searchAttributeDefinition: searchAttributeDefinition{
alias: alias,
field: datetimeField.field,
valueType: enumspb.INDEXED_VALUE_TYPE_DATETIME,
},
}
}
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.
// NewSearchAttributeInt creates a new integer search attribute given a predefined chasm field
func NewSearchAttributeInt(alias string, intField SearchAttributeFieldInt) SearchAttributeInt {
search_attribute.go
return SearchAttributeInt{
searchAttributeDefinition: searchAttributeDefinition{
alias: alias,
field: intField.field,
valueType: enumspb.INDEXED_VALUE_TYPE_INT,
},
}
}
// Value sets the integer value of the search attribute.
// NewSearchAttributeKeyword creates a new keyword search attribute given a predefined chasm field
func NewSearchAttributeKeyword(alias string, keywordField SearchAttributeFieldKeyword) SearchAttributeKeyword {
search_attribute.go
return SearchAttributeKeyword{
searchAttributeDefinition: searchAttributeDefinition{
alias: alias,
field: keywordField.field,
valueType: enumspb.INDEXED_VALUE_TYPE_KEYWORD,
},
}
}
func newSearchAttributeKeywordByField(field string) SearchAttributeKeyword {
search_attribute.go
return SearchAttributeKeyword{
searchAttributeDefinition: searchAttributeDefinition{
alias: field,
field: field,
valueType: enumspb.INDEXED_VALUE_TYPE_KEYWORD,
},
}
}
// Value sets the string value of the search attribute.
}
func newSearchAttributeKeywordListByField(field string) SearchAttributeKeywordList {
search_attribute.go
return SearchAttributeKeywordList{
searchAttributeDefinition: searchAttributeDefinition{
alias: field,
field: field,
valueType: enumspb.INDEXED_VALUE_TYPE_KEYWORD_LIST,
},
}
}
// Value sets the string list value of the search attribute.
func (*Predicate) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[0]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
func (*UniversalPredicateAttributes) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[1]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
func (*EmptyPredicateAttributes) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[2]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
func (*AndPredicateAttributes) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[3]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
func (*OrPredicateAttributes) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[4]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
func (*NotPredicateAttributes) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[5]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
func (*NamespaceIdPredicateAttributes) ProtoMessage() {}
func (x *NamespaceIdPredicateAttributes) ProtoReflect() protoreflect.Message {
predicates.pb.go
mi := &file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[6]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
func (*TaskTypePredicateAttributes) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[7]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
func (*DestinationPredicateAttributes) ProtoMessage() {}
func (x *DestinationPredicateAttributes) ProtoReflect() protoreflect.Message {
predicates.pb.go
mi := &file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[8]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
func (*OutboundTaskGroupPredicateAttributes) ProtoMessage() {}
func (x *OutboundTaskGroupPredicateAttributes) ProtoReflect() protoreflect.Message {
predicates.pb.go
mi := &file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[9]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
func (*OutboundTaskPredicateAttributes) ProtoMessage() {}
func (x *OutboundTaskPredicateAttributes) ProtoReflect() protoreflect.Message {
predicates.pb.go
mi := &file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[10]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
}
func init() { file_temporal_server_api_persistence_v1_predicates_proto_init() }
predicates.pb.go
func file_temporal_server_api_persistence_v1_predicates_proto_init() {
if File_temporal_server_api_persistence_v1_predicates_proto != nil {
return
}
file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[0].OneofWrappers = []any{
(*Predicate_UniversalPredicateAttributes)(nil),
(*Predicate_EmptyPredicateAttributes)(nil),
(*Predicate_AndPredicateAttributes)(nil),
(*Predicate_OrPredicateAttributes)(nil),
(*Predicate_NotPredicateAttributes)(nil),
(*Predicate_NamespaceIdPredicateAttributes)(nil),
(*Predicate_TaskTypePredicateAttributes)(nil),
(*Predicate_DestinationPredicateAttributes)(nil),
(*Predicate_OutboundTaskGroupPredicateAttributes)(nil),
(*Predicate_OutboundTaskPredicateAttributes)(nil),
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_persistence_v1_predicates_proto_rawDesc), len(file_temporal_server_api_persistence_v1_predicates_proto_rawDesc)),
NumEnums: 0,
NumMessages: 12,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_persistence_v1_predicates_proto_goTypes,
DependencyIndexes: file_temporal_server_api_persistence_v1_predicates_proto_depIdxs,
MessageInfos: file_temporal_server_api_persistence_v1_predicates_proto_msgTypes,
}.Build()
File_temporal_server_api_persistence_v1_predicates_proto = out.File
file_temporal_server_api_persistence_v1_predicates_proto_goTypes = nil
file_temporal_server_api_persistence_v1_predicates_proto_depIdxs = nil
}
)
func setUpPostgreSQLTest(t *testing.T, pluginName string, connectAttrs map[string]string) (PostgreSQLTestData, func()) {
postgresql_test_util.go
var testData PostgreSQLTestData
testData.Cfg = NewPostgreSQLConfig(pluginName, connectAttrs)
testData.Logger = log.NewZapLogger(zaptest.NewLogger(t))
mh := metricstest.NewCaptureHandler()
testData.Metrics = mh.StartCapture()
SetupPostgreSQLDatabase(t, testData.Cfg)
SetupPostgreSQLSchema(t, testData.Cfg)
testData.Factory = sql.NewFactory(
*testData.Cfg,
resolver.NewNoopResolver(),
testPostgreSQLClusterName,
testData.Logger,
mh,
serialization.NewSerializer(),
)
tearDown := func() {
testData.Factory.Close()
mh.StopCapture(testData.Metrics)
TearDownPostgreSQLDatabase(t, testData.Cfg)
}
}
// NewPostgreSQLConfig returns a new MySQL config for test
func NewPostgreSQLConfig(pluginName string, connectAttrs map[string]string) *config.SQL {
postgresql_test_util.go
return &config.SQL{
User: testPostgreSQLUser,
Password: testPostgreSQLPassword,
ConnectAddr: net.JoinHostPort(
environment.GetPostgreSQLAddress(),
strconv.Itoa(environment.GetPostgreSQLPort()),
),
ConnectProtocol: testPostgreSQLConnectionProtocol,
PluginName: pluginName,
DatabaseName: testPostgreSQLDatabaseNamePrefix + shuffle.String(testPostgreSQLDatabaseNameSuffix),
ConnectAttributes: connectAttrs,
}
}
adminCfg := *cfg
// NOTE need to connect with empty name to create new database
adminCfg.DatabaseName = ""
db, err := sql.NewSQLAdminDB(sqlplugin.DbKindUnknown, &adminCfg, resolver.NewNoopResolver(), log.NewTestLogger(), metrics.NoopMetricsHandler)
if err != nil {
t.Fatalf("unable to create PostgreSQL admin DB: %v", err)
}
if err != nil {
t.Fatalf("unable to create PostgreSQL database: %v", err)
}
}
db, err := sql.NewSQLAdminDB(sqlplugin.DbKindUnknown, cfg, resolver.NewNoopResolver(), log.NewTestLogger(), metrics.NoopMetricsHandler)
if err != nil {
t.Fatalf("unable to create PostgreSQL admin DB: %v", err)
}
if err != nil {
t.Fatal(err)
}
if err != nil {
t.Fatal(err)
}
if err = db.Exec(stmt); err != nil {
t.Fatal(err)
}
}
if err != nil {
t.Fatal(err)
}
if err != nil {
t.Fatal(err)
}
if err = db.Exec(stmt); err != nil {
t.Fatal(err)
}
}
adminCfg := *cfg
// NOTE need to connect with empty name to create new database
adminCfg.DatabaseName = ""
db, err := sql.NewSQLAdminDB(sqlplugin.DbKindUnknown, &adminCfg, resolver.NewNoopResolver(), log.NewTestLogger(), metrics.NoopMetricsHandler)
if err != nil {
t.Fatalf("unable to create PostgreSQL admin DB: %v", err)
}
if err != nil {
t.Fatalf("unable to drop PostgreSQL database: %v", err)
}
func (*HistoryTreeInfo) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_history_tree_proto_msgTypes[0]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
*x = HistoryBranch{}
mi := &file_temporal_server_api_persistence_v1_history_tree_proto_msgTypes[1]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *HistoryBranch) String() string {
func (*HistoryBranch) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_history_tree_proto_msgTypes[1]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
}
}
if x != nil {
return x.TreeId
}
return ""
}
if x != nil {
return x.BranchId
}
return ""
}
func (*HistoryBranchRange) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_history_tree_proto_msgTypes[2]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
}
if x != nil {
return x.BranchId
}
return ""
}
if x != nil {
return x.BeginNodeId
}
return 0
}
if x != nil {
return x.EndNodeId
}
return 0
}
}
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
}
metricsHandler metrics.Handler,
timeSource clock.TimeSource,
handle := &DatabaseHandle{
running: true,
connect: connect,
needsRefresh: needsRefresh,
metrics: metricsHandler,
logger: logger,
timeSource: timeSource,
}
handle.reporter = newDBMetricReporter(dbKind, handle)
handle.reporter.Start()
handle.reconnect(true)
return handle
}
// Close and reopen the underlying database connection
h.Lock()
defer h.Unlock()
// Don't reconnect if we've been closed
if !h.running {
return nil
}
if prevConn != nil {
if !force {
// Another goroutine already reconnected
}
now := h.timeSource.Now()
lastRefresh := h.lastRefresh
if now.Sub(lastRefresh) < sessionRefreshMinInternal {
h.logger.Warn("sql handle: did not refresh database connection pool because the last refresh was too close",
tag.Duration("min_refresh_interval_seconds", sessionRefreshMinInternal))
}
newConn, err := h.connect()
if err != nil {
h.logger.Error("sql handle: unable to refresh database connection pool", tag.Error(err))
handler := h.metrics.WithTags(metrics.FailureTag("error"))
}
return newConn
}
h.Lock()
defer h.Unlock()
if h.running {
h.running = false
if h.reporter != nil {
h.reporter.Stop()
}
db := h.db.Swap(nil)
if db != nil {
db.Close()
}
}
}
if db := h.db.Load(); db != nil {
}
if db := h.reconnect(false); db != nil {
}
if db := h.db.Load(); db != nil {
}
if db := h.reconnect(false); db != nil {
}
if h.needsRefresh(err) ||
errors.Is(err, driver.ErrBadConn) ||
errors.Is(err, io.ErrUnexpectedEOF) ||
errors.Is(err, io.EOF) ||
errors.Is(err, syscall.ECONNRESET) ||
errors.Is(err, syscall.ECONNABORTED) ||
errors.Is(err, syscall.ECONNREFUSED) {
h.reconnect(true)
return serviceerror.NewUnavailablef("database connection lost: %s", err.Error())
}
}
}
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
}
ctx context.Context,
row *sqlplugin.HistoryNodeRow,
// NOTE: txn_id is *= -1 within DB
row.TxnID = -row.TxnID
return pdb.NamedExecContext(ctx,
addHistoryNodesQuery,
row,
)
}
// DeleteFromHistoryNode delete a row from history_node table
ctx context.Context,
row *sqlplugin.HistoryNodeRow,
// NOTE: txn_id is *= -1 within DB
row.TxnID = -row.TxnID
return pdb.ExecContext(ctx,
deleteHistoryNodeQuery,
row.ShardID,
row.TreeID,
row.BranchID,
row.NodeID,
row.TxnID,
)
}
// SelectFromHistoryNode reads one or more rows from history_node table
ctx context.Context,
filter sqlplugin.HistoryNodeSelectFilter,
var query string
if filter.MetadataOnly {
query = getHistoryNodesReverseQuery
query = getHistoryNodesQuery
}
if filter.ReverseOrder {
args = []any{
filter.ShardID,
filter.PageSize,
}
args = []any{
filter.ShardID,
filter.TreeID,
filter.BranchID,
filter.MinNodeID,
-filter.MinTxnID, // NOTE: transaction ID is *= -1 when stored
filter.MinNodeID,
filter.MaxNodeID,
filter.PageSize,
}
}
err := pdb.SelectContext(ctx, &rows, query, args...)
if err != nil {
return nil, err
}
// NOTE: since we let txn_id multiple by -1 when inserting, we have to revert it back here
}
}
ctx context.Context,
row *sqlplugin.HistoryTreeRow,
return pdb.NamedExecContext(ctx,
addHistoryTreeQuery,
row,
)
}
// SelectFromHistoryTree reads one or more rows from history_tree table
// 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)
}
func caller(skip int) string {
}
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
}
}
tailTransactionID int64,
transactionIDToNode map[int64]historyNodeMetadata,
nodeIDToNodes := indexNodeIDToNode(transactionIDToNode)
nodeChain, err := reverselyLinkNode(
tailNodeID,
tailTransactionID,
transactionIDToNode,
)
if err != nil {
return nil, err
}
return nodesToTrim, nil
}
func indexNodeIDToNode(
transactionIDToNode map[int64]historyNodeMetadata,
// indexing node ID -> sorted nodeMetadata by transaction ID from high to low
nodeIDToNodes := make(map[int64][]historyNodeMetadata)
for _, node := range transactionIDToNode {
nodeMetadata := nodeIDToNodes[node.nodeID]
nodeMetadata = append(nodeMetadata, node)
nodeIDToNodes[node.nodeID] = nodeMetadata
}
for nodeID := range nodeIDToNodes {
nodes := nodeIDToNodes[nodeID]
// reverse sort by transaction ID
sort.Slice(nodes, func(i, j int) bool {
})
}
}
tailTransactionID int64,
transactionIDToNode map[int64]historyNodeMetadata,
// from tail node, trace back
transactionID := tailTransactionID
node, ok := transactionIDToNode[transactionID]
// sanity check node ID <-> transaction ID being unique
if !ok || node.nodeID != tailNodeID {
return nil, serviceerror.NewInternal(
fmt.Sprintf("unable to find or verify the tail history node, node ID: %v, transaction ID: %v",
}
nodes = append(nodes, node)
for node.nodeID > common.FirstEventID {
if prevNode, ok := transactionIDToNode[node.prevTransactionID]; !ok {
return nil, serviceerror.NewInternal(
fmt.Sprintf("unable to back trace history node, node ID: %v, transaction ID: %v, prev transaction ID: %v",
),
)
node = prevNode
nodes = append(nodes, node)
}
}
// node.nodeID == common.FirstEventID
// node.prevTransactionID == 0
return nil, serviceerror.NewInternal(
fmt.Sprintf("unable to back trace history node, node ID: %v, transaction ID: %v, prev transaction ID: %v",
nodeIDToNodes map[int64][]historyNodeMetadata,
nodeChain []historyNodeMetadata,
var nodesToTrim []historyNodeMetadata
// for each node on the chain, validate that the transaction ID being the largest
for _, validNode := range nodeChain {
for _, node := range nodeIDToNodes[validNode.nodeID] {
if node.transactionID <= validNode.transactionID {
break
}
}
}
}
}
c.recordingsLock.Lock()
defer c.recordingsLock.Unlock()
c.recordings[name] = append(c.recordings[name], r)
}
// CaptureHandler is a [metrics.Handler] that captures each metric recording.
// NewCaptureHandler creates a new [metrics.Handler] that captures.
return &CaptureHandler{
captures: map[*Capture]struct{}{},
capturesLock: &sync.RWMutex{},
captureCount: &atomic.Int32{},
}
}
// StartCapture returns a started capture. StopCapture should be called on
// complete.
capture := &Capture{recordings: make(CaptureSnapshot)}
c.capturesLock.Lock()
defer c.capturesLock.Unlock()
c.captures[capture] = struct{}{}
c.captureCount.Add(1)
return capture
}
// StopCapture stops capturing metrics for the given capture instance.
c.capturesLock.Lock()
defer c.capturesLock.Unlock()
delete(c.captures, capture)
c.captureCount.Add(-1)
}
// WithTags implements [metrics.Handler.WithTags].
return &CaptureHandler{
tags: append(append(make([]metrics.Tag, 0, len(c.tags)+len(tags)), c.tags...), tags...),
captures: c.captures,
capturesLock: c.capturesLock,
captureCount: c.captureCount,
}
}
func (c *CaptureHandler) record(name string, v any, unit metrics.MetricUnit, tags ...metrics.Tag) {
capture_handler.go
// If no captures are active, discard the metric to save memory.
if c.captureCount.Load() == 0 {
return
}
rec := &CapturedRecording{Value: v, Tags: make(map[string]string, len(c.tags)+len(tags)), Unit: unit}
capture_handler.go
for _, tag := range c.tags {
rec.Tags[tag.Key] = tag.Value
}
rec.Tags[tag.Key] = tag.Value
}
defer c.capturesLock.RUnlock()
for cap := range c.captures {
cap.record(name, rec)
}
}
// Counter implements [metrics.Handler.Counter].
return metrics.CounterFunc(func(v int64, tags ...metrics.Tag) { c.record(name, v, "", tags...) })
}
d driver.Driver,
resolver resolver.ServiceResolver,
db, err := createConnection(cfg, d, resolver)
if err != nil {
return nil, err
}
}
d driver.Driver,
resolver resolver.ServiceResolver,
var db *sqlx.DB
var err error
if cfg.PasswordCommand != nil {
db, err = d.CreateRefreshableConnection(func() (string, error) {
return buildDSN(cfg, resolver)
})
var dsn string
dsn, err = buildDSN(cfg, resolver)
if err != nil {
return nil, err
}
}
return nil, err
}
db.SetMaxOpenConns(cfg.MaxConns)
}
db.SetMaxIdleConns(cfg.MaxIdleConns)
}
db.SetConnMaxLifetime(cfg.MaxConnLifetime)
}
// Maps struct names in CamelCase to snake without need for db struct tags.
return db, nil
}
cfg *config.SQL,
r resolver.ServiceResolver,
password, err := cfg.ResolvePassword()
if err != nil {
return "", err
}
if err != nil {
return "", err
}
return fmt.Sprintf(
dsnFmt,
cfg.User,
url.QueryEscape(password),
resolvedAddr,
cfg.DatabaseName,
tlsAttrs.Encode(),
), nil
}
// nolint: revive
parameters := make(url.Values, len(cfg.ConnectAttributes))
for k, v := range cfg.ConnectAttributes {
value := strings.TrimSpace(v)
if parameters.Get(key) != "" {
panic(fmt.Sprintf("duplicate connection attr: %v:%v, %v:%v",
key,
))
}
}
if parameters.Get(sslMode) == "" {
if cfg.TLS.EnableHostVerification {
return nil, errors.New("failed to build postgresql DSN: sslkey connectAttribute is set but sslcert is not set")
}
}
}
}
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
}
}
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
}
tx *sqlx.Tx,
logger log.Logger,
mdb := &db{
dbKind: dbKind,
dbName: dbName,
dbDriver: dbDriver,
handle: handle,
tx: tx,
logger: logger,
}
mdb.converter = &converter{}
return mdb
}
if pdb.tx != nil {
}
}
// BeginTx starts a new transaction and returns a reference to the Tx object
db, err := pdb.handle.DB()
if err != nil {
// This error needs no conversion
return nil, err
}
if err != nil {
return nil, pdb.handle.ConvertError(err)
}
}
// Close closes the connection to the mysql db
pdb.handle.Close()
return nil
}
// PluginName returns the name of the mysql plugin
// Commit commits a previously started transaction
return pdb.tx.Commit()
}
// Rollback triggers rollback of a previously started transaction
// Helper methods to hide common error handling
func (pdb *db) ExecContext(ctx context.Context, stmt string, args ...any) (sql.Result, error) {
db.go
res, err := pdb.conn().ExecContext(ctx, stmt, args...)
return res, pdb.handle.ConvertError(err)
}
func (pdb *db) GetContext(ctx context.Context, dest any, query string, args ...any) error {
}
func (pdb *db) SelectContext(ctx context.Context, dest any, query string, args ...any) error {
db.go
err := pdb.conn().SelectContext(ctx, dest, query, args...)
return pdb.handle.ConvertError(err)
}
func (pdb *db) NamedExecContext(ctx context.Context, query string, arg any) (sql.Result, error) {
db.go
res, err := pdb.conn().NamedExecContext(ctx, query, arg)
return res, pdb.handle.ConvertError(err)
}
func (pdb *db) PrepareNamedContext(ctx context.Context, query string) (*sqlx.NamedStmt, error) {
metricsHandler metrics.Handler,
serializer serialization.Serializer,
return &Factory{
cfg: cfg,
clusterName: clusterName,
logger: logger,
serializer: serializer,
mainDBConn: NewRefCountedDBConn(sqlplugin.DbKindMain, &cfg, r, logger, metricsHandler),
}
}
// GetDB return a new SQL DB connection
// NewExecutionStore returns a new ExecutionStore
conn, err := f.mainDBConn.Get()
if err != nil {
return nil, err
}
}
// Close closes the factory
f.mainDBConn.ForceClose()
}
// NewRefCountedDBConn returns a logical mysql connection that
logger log.Logger,
metricsHandler metrics.Handler,
return DbConn{
dbKind: dbKind,
cfg: cfg,
resolver: r,
metrics: metricsHandler,
logger: logger,
}
}
// Get returns a db connection and increments a reference count.
// This method will create a new connection, if an existing connection
// does not exist
c.Lock()
defer c.Unlock()
if c.refCnt == 0 {
conn, err := NewSQLDB(c.dbKind, c.cfg, c.resolver, c.logger, c.metrics)
if err != nil {
return nil, err
}
}
return c, nil
}
// ForceClose ignores reference counts and shutsdown the underlying connection pool
c.Lock()
defer c.Unlock()
if c.DB != nil {
err := c.DB.Close()
if err != nil {
fmt.Println("failed to close database connection, may leak some connection", err)
}
}
}
// 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
// All retries are computed using the following formula:
// initialInterval * math.Pow(backoffCoefficient, currentAttempt)
func (p *ExponentialRetryPolicy) WithBackoffCoefficient(backoffCoefficient float64) *ExponentialRetryPolicy {
retrypolicy.go
p.backoffCoefficient = backoffCoefficient
return p
}
// WithMaximumInterval sets the maximum interval for each 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 {
)
// 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)
}
func (d gaugeDefinition) With(handler Handler) GaugeIface {
}
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{}
}
}
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 {
logger log.Logger,
metricsHandler metrics.Handler,
connect := func() (*sqlx.DB, error) {
if cfg.Connect != nil {
return cfg.Connect(cfg)
}
}
handle := sqlplugin.NewDatabaseHandle(dbKind, connect, needsRefresh, logger, metricsHandler, clock.NewRealTimeSource())
db := newDB(dbKind, cfg.DatabaseName, p.driver, handle, nil, logger)
return db, nil
}
cfg *config.SQL,
resolver resolver.ServiceResolver,
if cfg.DatabaseName != "" {
if err != nil {
return nil, err
}
}
// (which captures the config pointer) sees the correct DatabaseName on
// reconnect
var errors []error
for _, databaseName := range defaultDatabaseNames {
cfgCopy.DatabaseName = databaseName
if postgresqlSession, err := session.NewSession(
&cfgCopy,
p.driver,
resolver,
); err == nil {
return postgresqlSession.DB, nil
} else {
errors = append(errors, err)
}
}
func file_temporal_server_api_persistence_v1_chasm_proto_init() {
if File_temporal_server_api_persistence_v1_chasm_proto != nil {
return
}
file_temporal_server_api_persistence_v1_hsm_proto_init()
file_temporal_server_api_persistence_v1_chasm_proto_msgTypes[1].OneofWrappers = []any{
(*ChasmNodeMetadata_ComponentAttributes)(nil),
(*ChasmNodeMetadata_DataAttributes)(nil),
(*ChasmNodeMetadata_CollectionAttributes)(nil),
(*ChasmNodeMetadata_PointerAttributes)(nil),
}
file_temporal_server_api_persistence_v1_chasm_proto_msgTypes[10].OneofWrappers = []any{
(*ChasmNexusCompletion_Success)(nil),
(*ChasmNexusCompletion_Failure)(nil),
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_persistence_v1_chasm_proto_rawDesc), len(file_temporal_server_api_persistence_v1_chasm_proto_rawDesc)),
NumEnums: 0,
NumMessages: 15,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_persistence_v1_chasm_proto_goTypes,
DependencyIndexes: file_temporal_server_api_persistence_v1_chasm_proto_depIdxs,
MessageInfos: file_temporal_server_api_persistence_v1_chasm_proto_msgTypes,
}.Build()
File_temporal_server_api_persistence_v1_chasm_proto = out.File
file_temporal_server_api_persistence_v1_chasm_proto_goTypes = nil
file_temporal_server_api_persistence_v1_chasm_proto_depIdxs = nil
}
}
func 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
}
)
func NewHistoryBranchUtil(serializer serialization.Serializer) *HistoryBranchUtilImpl {
history_branch_util.go
return &HistoryBranchUtilImpl{
serializer: serializer,
}
}
func (u *HistoryBranchUtilImpl) NewHistoryBranch(
_ time.Duration, // executionTimeout
_ time.Duration, // retentionDuration
var id string
if branchID == nil {
id = primitives.NewUUID().String()
}
TreeId: treeID,
BranchId: id,
Ancestors: ancestors,
}
data, err := u.serializer.HistoryBranchToBlob(bi)
if err != nil {
return nil, err
}
}
func (u *HistoryBranchUtilImpl) ParseHistoryBranchInfo(
branchToken []byte,
return u.serializer.HistoryBranchFromBlob(branchToken)
}
func (u *HistoryBranchUtilImpl) UpdateHistoryBranchInfo(
branchInfo *persistencespb.HistoryBranch,
runID string,
bi, err := u.serializer.HistoryBranchFromBlob(branchToken)
if err != nil {
return nil, err
}
bi.BranchId = branchInfo.BranchId
bi.Ancestors = branchInfo.Ancestors
blob, err := u.serializer.HistoryBranchToBlob(bi)
if err != nil {
return nil, err
}
}
}
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:
}
func newDBMetricReporter(dbKind DbKind, handle *DatabaseHandle) *DBMetricsReporter {
db_metrics.go
reporter := &DBMetricsReporter{
interval: time.Minute,
handle: handle,
metrics: handle.metrics.WithTags(metrics.PersistenceDBKindTag(dbKind.String())),
quit: make(chan struct{}),
logger: handle.logger,
}
reporter.started.Store(false)
reporter.stopped.Store(false)
return reporter
}
// Start run metrics report in background
// yield control immediately without blocking
// safe to called multiple time
if !r.started.CompareAndSwap(false, true) {
return
}
go r.run()
}
defer r.wg.Done()
ticker := time.NewTicker(r.interval)
defer ticker.Stop()
for {
select {
return
case <-ticker.C:
if r.stopped.Load() {
// and wait for reporter to completely stopped
// safe to call multiple time
if r.stopped.CompareAndSwap(false, true) {
close(r.quit)
}
r.wg.Wait()
}
)
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
}
)
return &serializerImpl{encodingType: encodingTypeFromEnv()}
}
func (t *serializerImpl) EncodingType() enumspb.EncodingType {
}
func (t *serializerImpl) SerializeEvents(events []*historypb.HistoryEvent) (*commonpb.DataBlob, error) {
serializer.go
return t.serialize(&historypb.History{Events: events})
}
func (t *serializerImpl) DeserializeEvents(data *commonpb.DataBlob) ([]*historypb.HistoryEvent, error) {
serializer.go
if data == nil {
return nil, nil
}
return nil, nil
}
err := Decode(data, events)
if err != nil {
return nil, err
}
}
}
func (t *serializerImpl) serialize(p proto.Message) (*commonpb.DataBlob, error) {
serializer.go
if p == nil {
return nil, nil
}
if err != nil {
return nil, NewSerializationError(t.encodingType, err)
}
}
}
func (t *serializerImpl) HistoryTreeInfoToBlob(info *persistencespb.HistoryTreeInfo) (*commonpb.DataBlob, error) {
serializer.go
return encodeBlob(info, t.encodingType)
}
func (t *serializerImpl) HistoryTreeInfoFromBlob(data *commonpb.DataBlob) (*persistencespb.HistoryTreeInfo, error) {
}
func (t *serializerImpl) HistoryBranchToBlob(info *persistencespb.HistoryBranch) (*commonpb.DataBlob, error) {
serializer.go
return encodeBlob(info, t.encodingType)
}
// NOTE: HistoryBranch does not have an encoding type; so we use the serializer's encoding type.
func (t *serializerImpl) HistoryBranchFromBlob(data []byte) (*persistencespb.HistoryBranch, error) {
serializer.go
result := &persistencespb.HistoryBranch{}
return result, Decode(&commonpb.DataBlob{Data: data, EncodingType: t.encodingType}, result)
}
func (t *serializerImpl) WorkflowExecutionInfoToBlob(info *persistencespb.WorkflowExecutionInfo) (*commonpb.DataBlob, error) {
// 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
// CreateTaskReschedulePolicy creates a retry policy for rescheduling task with errors not equal to ErrTaskRetry
return backoff.NewExponentialRetryPolicy(taskRescheduleInitialInterval).
WithBackoffCoefficient(taskRescheduleBackoffCoefficient).
WithMaximumInterval(taskRescheduleMaxInterval).
WithExpirationInterval(backoff.NoInterval)
}
// CreateDependencyTaskNotCompletedReschedulePolicy creates a retry policy for rescheduling task with
// ErrDependencyTaskNotCompleted
return backoff.NewExponentialRetryPolicy(dependencyTaskNotCompletedRescheduleInitialInterval).
WithBackoffCoefficient(dependencyTaskNotCompletedRescheduleBackoffCoefficient).
WithMaximumInterval(dependencyTaskNotCompletedRescheduleMaxInterval).
WithExpirationInterval(backoff.NoInterval)
}
// CreateTaskNotReadyReschedulePolicy creates a retry policy for rescheduling task with ErrTaskRetry
return backoff.NewExponentialRetryPolicy(taskNotReadyRescheduleInitialInterval).
WithBackoffCoefficient(taskNotReadyRescheduleBackoffCoefficient).
WithMaximumInterval(taskNotReadyRescheduleMaxInterval).
WithExpirationInterval(backoff.NoInterval)
}
// CreateTaskResourceExhaustedReschedulePolicy creates a retry policy for rescheduling task with resource exhausted error
return backoff.NewExponentialRetryPolicy(taskResourceExhaustedRescheduleInitialInterval).
WithBackoffCoefficient(taskResourceExhaustedRescheduleBackoffCoefficient).
WithMaximumInterval(taskResourceExhaustedRescheduleMaxInterval).
WithExpirationInterval(backoff.NoInterval)
}
// CreateSdkClientFactoryRetryPolicy creates a retry policy to handle SdkClientFactory NewClient when frontend service is not ready
// 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
}
func init() { file_temporal_server_chasm_lib_scheduler_proto_v1_message_proto_init() }
message.pb.go
func file_temporal_server_chasm_lib_scheduler_proto_v1_message_proto_init() {
if File_temporal_server_chasm_lib_scheduler_proto_v1_message_proto != nil {
return
}
file_temporal_server_chasm_lib_scheduler_proto_v1_message_proto_msgTypes[4].OneofWrappers = []any{
(*BackfillerState_BackfillRequest)(nil),
(*BackfillerState_TriggerRequest)(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_scheduler_proto_v1_message_proto_rawDesc), len(file_temporal_server_chasm_lib_scheduler_proto_v1_message_proto_rawDesc)),
NumEnums: 0,
NumMessages: 12,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_chasm_lib_scheduler_proto_v1_message_proto_goTypes,
DependencyIndexes: file_temporal_server_chasm_lib_scheduler_proto_v1_message_proto_depIdxs,
MessageInfos: file_temporal_server_chasm_lib_scheduler_proto_v1_message_proto_msgTypes,
}.Build()
File_temporal_server_chasm_lib_scheduler_proto_v1_message_proto = out.File
file_temporal_server_chasm_lib_scheduler_proto_v1_message_proto_goTypes = nil
file_temporal_server_chasm_lib_scheduler_proto_v1_message_proto_depIdxs = nil
}
}
func init() { file_temporal_server_chasm_lib_activity_proto_v1_activity_state_proto_init() }
activity_state.pb.go
func file_temporal_server_chasm_lib_activity_proto_v1_activity_state_proto_init() {
if File_temporal_server_chasm_lib_activity_proto_v1_activity_state_proto != nil {
return
}
file_temporal_server_chasm_lib_activity_proto_v1_activity_state_proto_msgTypes[7].OneofWrappers = []any{
activity_state.pb.go
(*ActivityOutcome_Successful_)(nil),
(*ActivityOutcome_Failed_)(nil),
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_chasm_lib_activity_proto_v1_activity_state_proto_rawDesc), len(file_temporal_server_chasm_lib_activity_proto_v1_activity_state_proto_rawDesc)),
NumEnums: 2,
NumMessages: 11,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_chasm_lib_activity_proto_v1_activity_state_proto_goTypes,
DependencyIndexes: file_temporal_server_chasm_lib_activity_proto_v1_activity_state_proto_depIdxs,
EnumInfos: file_temporal_server_chasm_lib_activity_proto_v1_activity_state_proto_enumTypes,
MessageInfos: file_temporal_server_chasm_lib_activity_proto_v1_activity_state_proto_msgTypes,
}.Build()
File_temporal_server_chasm_lib_activity_proto_v1_activity_state_proto = out.File
file_temporal_server_chasm_lib_activity_proto_v1_activity_state_proto_goTypes = nil
file_temporal_server_chasm_lib_activity_proto_v1_activity_state_proto_depIdxs = nil
}
}
func init() { file_temporal_server_chasm_lib_nexusoperation_proto_v1_operation_proto_init() }
operation.pb.go
func file_temporal_server_chasm_lib_nexusoperation_proto_v1_operation_proto_init() {
if File_temporal_server_chasm_lib_nexusoperation_proto_v1_operation_proto != nil {
return
}
file_temporal_server_chasm_lib_nexusoperation_proto_v1_operation_proto_msgTypes[2].OneofWrappers = []any{
operation.pb.go
(*OperationOutcome_Successful_)(nil),
(*OperationOutcome_Failed_)(nil),
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_chasm_lib_nexusoperation_proto_v1_operation_proto_rawDesc), len(file_temporal_server_chasm_lib_nexusoperation_proto_v1_operation_proto_rawDesc)),
NumEnums: 2,
NumMessages: 8,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_chasm_lib_nexusoperation_proto_v1_operation_proto_goTypes,
DependencyIndexes: file_temporal_server_chasm_lib_nexusoperation_proto_v1_operation_proto_depIdxs,
EnumInfos: file_temporal_server_chasm_lib_nexusoperation_proto_v1_operation_proto_enumTypes,
MessageInfos: file_temporal_server_chasm_lib_nexusoperation_proto_v1_operation_proto_msgTypes,
}.Build()
File_temporal_server_chasm_lib_nexusoperation_proto_v1_operation_proto = out.File
file_temporal_server_chasm_lib_nexusoperation_proto_v1_operation_proto_goTypes = nil
file_temporal_server_chasm_lib_nexusoperation_proto_v1_operation_proto_depIdxs = nil
}
// NewRoute returns a new [Route] instance with the given components.
return Route[T]{components: components}
}
// RouteBuilder is a builder for the [Route] interface.
// NewBuilder creates a new [RouteBuilder] instance, which can be used to define a new [Route] via a fluent API.
return &RouteBuilder[T]{}
}
// With adds a series of [Component] instances to the [Route].
r.components = append(r.components, c...)
return r
}
// Constant adds a [Constant] component to the [Route].
return r.With(Constant[T](values...))
}
// StringVariable adds a [StringVariable] component to the [Route].
func (r *RouteBuilder[T]) StringVariable(name string, getter func(*T) *string) *RouteBuilder[T] {
route.go
return r.With(StringVariable[T](name, getter))
}
// Build returns a read-only [Route].
return NewRoute[T](r.components...)
}
// Representation returns the [github.com/gorilla/mux] compatible string representation of the route for usage in a
// Constant returns a [Component] that represents a series of constant HTTP path components in a Route.
// They will be joined via strings when used to construct a path or path representation.
return values
}
type constant[T any] []string
// StringVariable returns a [Component] that represents a string variable in a Route.
return stringVariable[T]{name, getter}
}
type stringVariable[T any] struct {
)
// lookup localhost and favor the first ipv4 address
// unless there are only ipv6 addresses available
ips, err := net.LookupIP(domain)
if err != nil || len(ips) == 0 {
// fallback to default instead of error
return localhostIPDefault
}
if ip4 := ip.To4(); ip4 != nil {
}
}
return ips[len(ips)-1].String()
// GetLocalhostIP returns the ip address of the localhost domain
localhostIP := os.Getenv(localhostIPEnv)
ip := net.ParseIP(localhostIP)
if ip != nil {
// if localhost is an ip return it
return ip.String()
}
// otherwise, ignore the value and lookup `localhost`
}
// GetPostgreSQLAddress return the Postgres address
addr := os.Getenv(postgresSeedsEnv)
if addr == "" {
addr = GetLocalhostIP()
}
return addr
}
// GetPostgreSQLPort return the Postgres port
port := os.Getenv(postgresPortEnv)
if port == "" {
return postgresDefaultPort
}
p, err := strconv.Atoi(port)
if err != nil {
}
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 file_temporal_server_api_schedule_v1_message_proto_init() {
if File_temporal_server_api_schedule_v1_message_proto != nil {
return
}
file_temporal_server_api_schedule_v1_message_proto_msgTypes[7].OneofWrappers = []any{
message.pb.go
(*WatchWorkflowResponse_Result)(nil),
(*WatchWorkflowResponse_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_schedule_v1_message_proto_rawDesc), len(file_temporal_server_api_schedule_v1_message_proto_rawDesc)),
NumEnums: 0,
NumMessages: 13,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_schedule_v1_message_proto_goTypes,
DependencyIndexes: file_temporal_server_api_schedule_v1_message_proto_depIdxs,
MessageInfos: file_temporal_server_api_schedule_v1_message_proto_msgTypes,
}.Build()
File_temporal_server_api_schedule_v1_message_proto = out.File
file_temporal_server_api_schedule_v1_message_proto_goTypes = nil
file_temporal_server_api_schedule_v1_message_proto_depIdxs = nil
}
}
func init() { file_temporal_server_chasm_lib_callback_proto_v1_message_proto_init() }
message.pb.go
func file_temporal_server_chasm_lib_callback_proto_v1_message_proto_init() {
if File_temporal_server_chasm_lib_callback_proto_v1_message_proto != nil {
return
}
file_temporal_server_chasm_lib_callback_proto_v1_message_proto_msgTypes[1].OneofWrappers = []any{
message.pb.go
(*Callback_Nexus_)(nil),
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_chasm_lib_callback_proto_v1_message_proto_rawDesc), len(file_temporal_server_chasm_lib_callback_proto_v1_message_proto_rawDesc)),
NumEnums: 1,
NumMessages: 5,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_chasm_lib_callback_proto_v1_message_proto_goTypes,
DependencyIndexes: file_temporal_server_chasm_lib_callback_proto_v1_message_proto_depIdxs,
EnumInfos: file_temporal_server_chasm_lib_callback_proto_v1_message_proto_enumTypes,
MessageInfos: file_temporal_server_chasm_lib_callback_proto_v1_message_proto_msgTypes,
}.Build()
File_temporal_server_chasm_lib_callback_proto_v1_message_proto = out.File
file_temporal_server_chasm_lib_callback_proto_v1_message_proto_goTypes = nil
file_temporal_server_chasm_lib_callback_proto_v1_message_proto_depIdxs = nil
}
// 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
}
}
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
}
}
func init() { file_temporal_server_chasm_lib_scheduler_proto_v1_request_response_proto_init() }
request_response.pb.go
func file_temporal_server_chasm_lib_scheduler_proto_v1_request_response_proto_init() {
if File_temporal_server_chasm_lib_scheduler_proto_v1_request_response_proto != nil {
return
}
file_temporal_server_chasm_lib_scheduler_proto_v1_message_proto_init()
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_chasm_lib_scheduler_proto_v1_request_response_proto_rawDesc), len(file_temporal_server_chasm_lib_scheduler_proto_v1_request_response_proto_rawDesc)),
NumEnums: 0,
NumMessages: 18,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_chasm_lib_scheduler_proto_v1_request_response_proto_goTypes,
DependencyIndexes: file_temporal_server_chasm_lib_scheduler_proto_v1_request_response_proto_depIdxs,
MessageInfos: file_temporal_server_chasm_lib_scheduler_proto_v1_request_response_proto_msgTypes,
}.Build()
File_temporal_server_chasm_lib_scheduler_proto_v1_request_response_proto = out.File
file_temporal_server_chasm_lib_scheduler_proto_v1_request_response_proto_goTypes = nil
file_temporal_server_chasm_lib_scheduler_proto_v1_request_response_proto_depIdxs = nil
}
resolver ReplicationResolver,
mutations ...Mutation,
if resolver == nil {
return nil, serviceerror.NewInvalidArgument("replicationResolver must be provided")
}
info: detail.Info,
config: detail.Config,
configVersion: detail.ConfigVersion,
customSearchAttributesMapper: CustomSearchAttributesMapper{
fieldToAlias: detail.Config.CustomSearchAttributeAliases,
aliasToField: util.InverseMap(detail.Config.CustomSearchAttributeAliases),
},
replicationResolver: resolver,
}
for _, m := range mutations {
}
}
}
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_activity_proto_v1_request_response_proto_init() }
request_response.pb.go
func file_temporal_server_chasm_lib_activity_proto_v1_request_response_proto_init() {
if File_temporal_server_chasm_lib_activity_proto_v1_request_response_proto != nil {
return
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_chasm_lib_activity_proto_v1_request_response_proto_rawDesc), len(file_temporal_server_chasm_lib_activity_proto_v1_request_response_proto_rawDesc)),
NumEnums: 0,
NumMessages: 20,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_chasm_lib_activity_proto_v1_request_response_proto_goTypes,
DependencyIndexes: file_temporal_server_chasm_lib_activity_proto_v1_request_response_proto_depIdxs,
MessageInfos: file_temporal_server_chasm_lib_activity_proto_v1_request_response_proto_msgTypes,
}.Build()
File_temporal_server_chasm_lib_activity_proto_v1_request_response_proto = out.File
file_temporal_server_chasm_lib_activity_proto_v1_request_response_proto_goTypes = nil
file_temporal_server_chasm_lib_activity_proto_v1_request_response_proto_depIdxs = nil
}
}
func init() { file_temporal_server_chasm_lib_nexusoperation_proto_v1_request_response_proto_init() }
request_response.pb.go
func file_temporal_server_chasm_lib_nexusoperation_proto_v1_request_response_proto_init() {
if File_temporal_server_chasm_lib_nexusoperation_proto_v1_request_response_proto != nil {
return
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_chasm_lib_nexusoperation_proto_v1_request_response_proto_rawDesc), len(file_temporal_server_chasm_lib_nexusoperation_proto_v1_request_response_proto_rawDesc)),
NumEnums: 0,
NumMessages: 12,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_chasm_lib_nexusoperation_proto_v1_request_response_proto_goTypes,
DependencyIndexes: file_temporal_server_chasm_lib_nexusoperation_proto_v1_request_response_proto_depIdxs,
MessageInfos: file_temporal_server_chasm_lib_nexusoperation_proto_v1_request_response_proto_msgTypes,
}.Build()
File_temporal_server_chasm_lib_nexusoperation_proto_v1_request_response_proto = out.File
file_temporal_server_chasm_lib_nexusoperation_proto_v1_request_response_proto_goTypes = nil
file_temporal_server_chasm_lib_nexusoperation_proto_v1_request_response_proto_depIdxs = nil
}
}
func init() { file_temporal_server_chasm_lib_tests_proto_v1_request_response_proto_init() }
request_response.pb.go
func file_temporal_server_chasm_lib_tests_proto_v1_request_response_proto_init() {
if File_temporal_server_chasm_lib_tests_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_tests_proto_v1_request_response_proto_rawDesc), len(file_temporal_server_chasm_lib_tests_proto_v1_request_response_proto_rawDesc)),
NumEnums: 0,
NumMessages: 2,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_chasm_lib_tests_proto_v1_request_response_proto_goTypes,
DependencyIndexes: file_temporal_server_chasm_lib_tests_proto_v1_request_response_proto_depIdxs,
MessageInfos: file_temporal_server_chasm_lib_tests_proto_v1_request_response_proto_msgTypes,
}.Build()
File_temporal_server_chasm_lib_tests_proto_v1_request_response_proto = out.File
file_temporal_server_chasm_lib_tests_proto_v1_request_response_proto_goTypes = nil
file_temporal_server_chasm_lib_tests_proto_v1_request_response_proto_depIdxs = nil
}
// newJSONHistoryTokenSerializer creates a new instance of TaskTokenSerializer
func newJSONHistoryTokenSerializer() *jsonHistoryTokenSerializer {
json_history_token_serializer.go
return &jsonHistoryTokenSerializer{}
}
func (t *historyPagingToken) SetRangeIndexes(
func (j *jsonHistoryTokenSerializer) Serialize(
token *historyPagingToken,
data, err := json.Marshal(token)
return data, err
}
func (j *jsonHistoryTokenSerializer) Deserialize(
defaultLastNodeID int64,
defaultLastTransactionID int64,
if len(data) == 0 {
token := historyPagingToken{
LastEventID: defaultLastEventID,
CurrentRangeIndex: notStartedIndex,
LastNodeID: defaultLastNodeID,
LastTransactionID: defaultLastTransactionID,
}
return &token, nil
}
err := json.Unmarshal(data, &token)
return &token, err
}
}
func init() { file_temporal_server_api_persistence_v1_task_queues_proto_init() }
task_queues.pb.go
func file_temporal_server_api_persistence_v1_task_queues_proto_init() {
if File_temporal_server_api_persistence_v1_task_queues_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_persistence_v1_task_queues_proto_rawDesc), len(file_temporal_server_api_persistence_v1_task_queues_proto_rawDesc)),
NumEnums: 1,
NumMessages: 13,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_persistence_v1_task_queues_proto_goTypes,
DependencyIndexes: file_temporal_server_api_persistence_v1_task_queues_proto_depIdxs,
EnumInfos: file_temporal_server_api_persistence_v1_task_queues_proto_enumTypes,
MessageInfos: file_temporal_server_api_persistence_v1_task_queues_proto_msgTypes,
}.Build()
File_temporal_server_api_persistence_v1_task_queues_proto = out.File
file_temporal_server_api_persistence_v1_task_queues_proto_goTypes = nil
file_temporal_server_api_persistence_v1_task_queues_proto_depIdxs = nil
}
}
func file_temporal_server_api_routing_v1_extension_proto_init() {
if File_temporal_server_api_routing_v1_extension_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_routing_v1_extension_proto_rawDesc), len(file_temporal_server_api_routing_v1_extension_proto_rawDesc)),
NumEnums: 0,
NumMessages: 1,
NumExtensions: 1,
NumServices: 0,
},
GoTypes: file_temporal_server_api_routing_v1_extension_proto_goTypes,
DependencyIndexes: file_temporal_server_api_routing_v1_extension_proto_depIdxs,
MessageInfos: file_temporal_server_api_routing_v1_extension_proto_msgTypes,
ExtensionInfos: file_temporal_server_api_routing_v1_extension_proto_extTypes,
}.Build()
File_temporal_server_api_routing_v1_extension_proto = out.File
file_temporal_server_api_routing_v1_extension_proto_goTypes = nil
file_temporal_server_api_routing_v1_extension_proto_depIdxs = nil
}
}
func init() { file_temporal_server_chasm_lib_activity_proto_v1_tasks_proto_init() }
tasks.pb.go
func file_temporal_server_chasm_lib_activity_proto_v1_tasks_proto_init() {
if File_temporal_server_chasm_lib_activity_proto_v1_tasks_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_chasm_lib_activity_proto_v1_tasks_proto_rawDesc), len(file_temporal_server_chasm_lib_activity_proto_v1_tasks_proto_rawDesc)),
NumEnums: 2,
NumMessages: 5,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_chasm_lib_activity_proto_v1_tasks_proto_goTypes,
DependencyIndexes: file_temporal_server_chasm_lib_activity_proto_v1_tasks_proto_depIdxs,
EnumInfos: file_temporal_server_chasm_lib_activity_proto_v1_tasks_proto_enumTypes,
MessageInfos: file_temporal_server_chasm_lib_activity_proto_v1_tasks_proto_msgTypes,
}.Build()
File_temporal_server_chasm_lib_activity_proto_v1_tasks_proto = out.File
file_temporal_server_chasm_lib_activity_proto_v1_tasks_proto_goTypes = nil
file_temporal_server_chasm_lib_activity_proto_v1_tasks_proto_depIdxs = nil
}
}
func NewSQLStore(db sqlplugin.DB, logger log.Logger, serializer serialization.Serializer) SqlStore {
common.go
return SqlStore{
DB: db,
logger: logger,
serializer: serializer,
}
}
func (m *SqlStore) GetName() string {
}
func (m *SqlStore) txExecute(ctx context.Context, operation string, f func(tx sqlplugin.Tx) error) error {
common.go
tx, err := m.DB.BeginTx(ctx)
if err != nil {
return serviceerror.NewUnavailablef("%s failed. Failed to start transaction. Error: %v", operation, err)
}
if err != nil {
rollBackErr := tx.Rollback()
if rollBackErr != nil {
}
return json.Marshal(token)
}
var token T
if err := json.Unmarshal(payload, &token); err != nil {
return nil, err
}
}
// RegisterPlugin will register a SQL plugin
if _, ok := supportedPlugins[pluginName]; ok {
panic("plugin " + pluginName + " already registered")
}
}
logger log.Logger,
mh metrics.Handler,
return createDB[sqlplugin.DB](dbKind, cfg, r, logger, mh)
}
// NewSQLAdminDB returns a AdminDB.
logger log.Logger,
mh metrics.Handler,
return createDB[sqlplugin.AdminDB](dbKind, cfg, r, logger, mh)
}
func createDB[T any](
logger log.Logger,
mh metrics.Handler,
var res T
plugin, err := getPlugin(cfg.PluginName)
if err != nil {
return res, err
}
if err != nil {
return res, err
}
//revive:disable-next-line:unchecked-type-assertion
return res, err
}
plugin, ok := supportedPlugins[pluginName]
if !ok {
keys := expmaps.Keys(supportedPlugins)
slices.Sort(keys)
}
func file_temporal_server_api_adminservice_v1_service_proto_init() {
if File_temporal_server_api_adminservice_v1_service_proto != nil {
return
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_adminservice_v1_service_proto_rawDesc), len(file_temporal_server_api_adminservice_v1_service_proto_rawDesc)),
NumEnums: 0,
NumMessages: 0,
NumExtensions: 0,
NumServices: 1,
},
GoTypes: file_temporal_server_api_adminservice_v1_service_proto_goTypes,
DependencyIndexes: file_temporal_server_api_adminservice_v1_service_proto_depIdxs,
}.Build()
File_temporal_server_api_adminservice_v1_service_proto = out.File
file_temporal_server_api_adminservice_v1_service_proto_goTypes = nil
file_temporal_server_api_adminservice_v1_service_proto_depIdxs = nil
}
}
func file_temporal_server_api_archiver_v1_message_proto_init() {
if File_temporal_server_api_archiver_v1_message_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_archiver_v1_message_proto_rawDesc), len(file_temporal_server_api_archiver_v1_message_proto_rawDesc)),
NumEnums: 0,
NumMessages: 4,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_archiver_v1_message_proto_goTypes,
DependencyIndexes: file_temporal_server_api_archiver_v1_message_proto_depIdxs,
MessageInfos: file_temporal_server_api_archiver_v1_message_proto_msgTypes,
}.Build()
File_temporal_server_api_archiver_v1_message_proto = out.File
file_temporal_server_api_archiver_v1_message_proto_goTypes = nil
file_temporal_server_api_archiver_v1_message_proto_depIdxs = nil
}
}
func file_temporal_server_api_chasm_v1_message_proto_init() {
if File_temporal_server_api_chasm_v1_message_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_chasm_v1_message_proto_rawDesc), len(file_temporal_server_api_chasm_v1_message_proto_rawDesc)),
NumEnums: 0,
NumMessages: 1,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_chasm_v1_message_proto_goTypes,
DependencyIndexes: file_temporal_server_api_chasm_v1_message_proto_depIdxs,
MessageInfos: file_temporal_server_api_chasm_v1_message_proto_msgTypes,
}.Build()
File_temporal_server_api_chasm_v1_message_proto = out.File
file_temporal_server_api_chasm_v1_message_proto_goTypes = nil
file_temporal_server_api_chasm_v1_message_proto_depIdxs = nil
}
}
func file_temporal_server_api_checksum_v1_message_proto_init() {
if File_temporal_server_api_checksum_v1_message_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_checksum_v1_message_proto_rawDesc), len(file_temporal_server_api_checksum_v1_message_proto_rawDesc)),
NumEnums: 0,
NumMessages: 1,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_checksum_v1_message_proto_goTypes,
DependencyIndexes: file_temporal_server_api_checksum_v1_message_proto_depIdxs,
MessageInfos: file_temporal_server_api_checksum_v1_message_proto_msgTypes,
}.Build()
File_temporal_server_api_checksum_v1_message_proto = out.File
file_temporal_server_api_checksum_v1_message_proto_goTypes = nil
file_temporal_server_api_checksum_v1_message_proto_depIdxs = nil
}
}
func file_temporal_server_api_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_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_activity_proto_v1_service_proto_init() }
service.pb.go
func file_temporal_server_chasm_lib_activity_proto_v1_service_proto_init() {
if File_temporal_server_chasm_lib_activity_proto_v1_service_proto != nil {
return
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_chasm_lib_activity_proto_v1_service_proto_rawDesc), len(file_temporal_server_chasm_lib_activity_proto_v1_service_proto_rawDesc)),
NumEnums: 0,
NumMessages: 0,
NumExtensions: 0,
NumServices: 1,
},
GoTypes: file_temporal_server_chasm_lib_activity_proto_v1_service_proto_goTypes,
DependencyIndexes: file_temporal_server_chasm_lib_activity_proto_v1_service_proto_depIdxs,
}.Build()
File_temporal_server_chasm_lib_activity_proto_v1_service_proto = out.File
file_temporal_server_chasm_lib_activity_proto_v1_service_proto_goTypes = nil
file_temporal_server_chasm_lib_activity_proto_v1_service_proto_depIdxs = nil
}
}
func init() { file_temporal_server_chasm_lib_callback_proto_v1_tasks_proto_init() }
tasks.pb.go
func file_temporal_server_chasm_lib_callback_proto_v1_tasks_proto_init() {
if File_temporal_server_chasm_lib_callback_proto_v1_tasks_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_chasm_lib_callback_proto_v1_tasks_proto_rawDesc), len(file_temporal_server_chasm_lib_callback_proto_v1_tasks_proto_rawDesc)),
NumEnums: 0,
NumMessages: 2,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_chasm_lib_callback_proto_v1_tasks_proto_goTypes,
DependencyIndexes: file_temporal_server_chasm_lib_callback_proto_v1_tasks_proto_depIdxs,
MessageInfos: file_temporal_server_chasm_lib_callback_proto_v1_tasks_proto_msgTypes,
}.Build()
File_temporal_server_chasm_lib_callback_proto_v1_tasks_proto = out.File
file_temporal_server_chasm_lib_callback_proto_v1_tasks_proto_goTypes = nil
file_temporal_server_chasm_lib_callback_proto_v1_tasks_proto_depIdxs = nil
}
}
func init() { file_temporal_server_chasm_lib_nexusoperation_proto_v1_service_proto_init() }
service.pb.go
func file_temporal_server_chasm_lib_nexusoperation_proto_v1_service_proto_init() {
if File_temporal_server_chasm_lib_nexusoperation_proto_v1_service_proto != nil {
return
}
file_temporal_server_chasm_lib_nexusoperation_proto_v1_request_response_proto_init()
service.pb.go
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_chasm_lib_nexusoperation_proto_v1_service_proto_rawDesc), len(file_temporal_server_chasm_lib_nexusoperation_proto_v1_service_proto_rawDesc)),
NumEnums: 0,
NumMessages: 0,
NumExtensions: 0,
NumServices: 1,
},
GoTypes: file_temporal_server_chasm_lib_nexusoperation_proto_v1_service_proto_goTypes,
DependencyIndexes: file_temporal_server_chasm_lib_nexusoperation_proto_v1_service_proto_depIdxs,
}.Build()
File_temporal_server_chasm_lib_nexusoperation_proto_v1_service_proto = out.File
file_temporal_server_chasm_lib_nexusoperation_proto_v1_service_proto_goTypes = nil
file_temporal_server_chasm_lib_nexusoperation_proto_v1_service_proto_depIdxs = nil
}
}
func init() { file_temporal_server_chasm_lib_nexusoperation_proto_v1_tasks_proto_init() }
tasks.pb.go
func file_temporal_server_chasm_lib_nexusoperation_proto_v1_tasks_proto_init() {
if File_temporal_server_chasm_lib_nexusoperation_proto_v1_tasks_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_chasm_lib_nexusoperation_proto_v1_tasks_proto_rawDesc), len(file_temporal_server_chasm_lib_nexusoperation_proto_v1_tasks_proto_rawDesc)),
NumEnums: 0,
NumMessages: 7,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_chasm_lib_nexusoperation_proto_v1_tasks_proto_goTypes,
DependencyIndexes: file_temporal_server_chasm_lib_nexusoperation_proto_v1_tasks_proto_depIdxs,
MessageInfos: file_temporal_server_chasm_lib_nexusoperation_proto_v1_tasks_proto_msgTypes,
}.Build()
File_temporal_server_chasm_lib_nexusoperation_proto_v1_tasks_proto = out.File
file_temporal_server_chasm_lib_nexusoperation_proto_v1_tasks_proto_goTypes = nil
file_temporal_server_chasm_lib_nexusoperation_proto_v1_tasks_proto_depIdxs = nil
}
}
func init() { file_temporal_server_chasm_lib_scheduler_proto_v1_service_proto_init() }
service.pb.go
func file_temporal_server_chasm_lib_scheduler_proto_v1_service_proto_init() {
if File_temporal_server_chasm_lib_scheduler_proto_v1_service_proto != nil {
return
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_chasm_lib_scheduler_proto_v1_service_proto_rawDesc), len(file_temporal_server_chasm_lib_scheduler_proto_v1_service_proto_rawDesc)),
NumEnums: 0,
NumMessages: 0,
NumExtensions: 0,
NumServices: 1,
},
GoTypes: file_temporal_server_chasm_lib_scheduler_proto_v1_service_proto_goTypes,
DependencyIndexes: file_temporal_server_chasm_lib_scheduler_proto_v1_service_proto_depIdxs,
}.Build()
File_temporal_server_chasm_lib_scheduler_proto_v1_service_proto = out.File
file_temporal_server_chasm_lib_scheduler_proto_v1_service_proto_goTypes = nil
file_temporal_server_chasm_lib_scheduler_proto_v1_service_proto_depIdxs = nil
}
}
func init() { file_temporal_server_chasm_lib_scheduler_proto_v1_tasks_proto_init() }
tasks.pb.go
func file_temporal_server_chasm_lib_scheduler_proto_v1_tasks_proto_init() {
if File_temporal_server_chasm_lib_scheduler_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_scheduler_proto_v1_tasks_proto_rawDesc), len(file_temporal_server_chasm_lib_scheduler_proto_v1_tasks_proto_rawDesc)),
NumEnums: 0,
NumMessages: 7,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_chasm_lib_scheduler_proto_v1_tasks_proto_goTypes,
DependencyIndexes: file_temporal_server_chasm_lib_scheduler_proto_v1_tasks_proto_depIdxs,
MessageInfos: file_temporal_server_chasm_lib_scheduler_proto_v1_tasks_proto_msgTypes,
}.Build()
File_temporal_server_chasm_lib_scheduler_proto_v1_tasks_proto = out.File
file_temporal_server_chasm_lib_scheduler_proto_v1_tasks_proto_goTypes = nil
file_temporal_server_chasm_lib_scheduler_proto_v1_tasks_proto_depIdxs = nil
}
}
func init() { file_temporal_server_chasm_lib_tests_proto_v1_message_proto_init() }
message.pb.go
func file_temporal_server_chasm_lib_tests_proto_v1_message_proto_init() {
if File_temporal_server_chasm_lib_tests_proto_v1_message_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_chasm_lib_tests_proto_v1_message_proto_rawDesc), len(file_temporal_server_chasm_lib_tests_proto_v1_message_proto_rawDesc)),
NumEnums: 0,
NumMessages: 4,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_chasm_lib_tests_proto_v1_message_proto_goTypes,
DependencyIndexes: file_temporal_server_chasm_lib_tests_proto_v1_message_proto_depIdxs,
MessageInfos: file_temporal_server_chasm_lib_tests_proto_v1_message_proto_msgTypes,
}.Build()
File_temporal_server_chasm_lib_tests_proto_v1_message_proto = out.File
file_temporal_server_chasm_lib_tests_proto_v1_message_proto_goTypes = nil
file_temporal_server_chasm_lib_tests_proto_v1_message_proto_depIdxs = nil
}
}
func init() { file_temporal_server_chasm_lib_tests_proto_v1_service_proto_init() }
service.pb.go
func file_temporal_server_chasm_lib_tests_proto_v1_service_proto_init() {
if File_temporal_server_chasm_lib_tests_proto_v1_service_proto != nil {
return
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_chasm_lib_tests_proto_v1_service_proto_rawDesc), len(file_temporal_server_chasm_lib_tests_proto_v1_service_proto_rawDesc)),
NumEnums: 0,
NumMessages: 0,
NumExtensions: 0,
NumServices: 1,
},
GoTypes: file_temporal_server_chasm_lib_tests_proto_v1_service_proto_goTypes,
DependencyIndexes: file_temporal_server_chasm_lib_tests_proto_v1_service_proto_depIdxs,
}.Build()
File_temporal_server_chasm_lib_tests_proto_v1_service_proto = out.File
file_temporal_server_chasm_lib_tests_proto_v1_service_proto_goTypes = nil
file_temporal_server_chasm_lib_tests_proto_v1_service_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.
}
}
const pgxDriverName = "pgx"
cfg, err := pgx.ParseConfig(dsn)
if err != nil {
return nil, err
}
db := sqlx.NewDb(sql.OpenDB(connector), pgxDriverName)
if err := db.Ping(); err != nil {
_ = db.Close()
return nil, err
}
}
// rejected. Encoding time.Duration as int8 (bigint in PostgreSQL) instead
// results in the integer nanosecond count
// pgx tracks the value type and the pointer type separately, so
// both must be remapped.
conn.TypeMap().RegisterDefaultPgType(time.Duration(0), "int8")
conn.TypeMap().RegisterDefaultPgType((*time.Duration)(nil), "int8")
return nil
}
func (p *PGXDriver) IsDupEntryError(err error) bool {
}
pqErr, ok := err.(*pgconn.PgError)
if !ok {
return false
}
return isConnNeedsRefreshError(pqErr.Code, pqErr.Message)
}
// treat the fields independently), or the zero value of its type (if you want to treat the fields
// as a group and default unset fields to zero).
return func(v any) (T, error) {
// if we already have the right type, no conversion is necessary
if typedV, ok := v.(T); ok {
return typedV, nil
}
// Deep-copy the default and decode over it. This allows using e.g. a struct with some
// default fields filled in and a config that only set some fields.
dec, err := mapstructure.NewDecoder(&mapstructure.DecoderConfig{
Result: &out,
DecodeHook: mapstructure.ComposeDecodeHookFunc(
mapstructureHookDuration,
mapstructureHookTimestamp,
mapstructureHookProtoEnum,
mapstructureHookGeneric,
),
})
if err != nil {
return out, err
}
return out, err
}
}
}
func 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
)
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]
transactionSizeLimit dynamicconfig.IntPropertyFn,
enableBestEffortDeleteTasksOnWorkflowUpdate dynamicconfig.BoolPropertyFn,
return &executionManagerImpl{
serializer: serializer,
eventBlobCache: eventBlobCache,
persistence: persistence,
logger: logger,
pagingTokenSerializer: newJSONHistoryTokenSerializer(),
transactionSizeLimit: transactionSizeLimit,
enableBestEffortDeleteTasksOnWorkflowUpdate: enableBestEffortDeleteTasksOnWorkflowUpdate,
}
}
func (m *executionManagerImpl) GetName() string {
}
return m.persistence.GetHistoryBranchUtil()
}
// historySizeRollback records HistorySize increments applied to caller-owned ExecutionStats
)
if path := hasSharedStructure(reflect.ValueOf(def), "root"); path != "" {
sharedStructureWarnings.Store(key, path)
}
}
// 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
}
// GetBeginNodeID gets node id from last ancestor
if len(bi.Ancestors) == 0 {
// root branch
return 1
}
return bi.Ancestors[idx].GetEndNodeId()
}
if len(ans) > 0 {
sort.Slice(ans, func(i, j int) bool { return (ans)[i].GetEndNodeId() < (ans)[j].GetEndNodeId() })
(ans)[0].BeginNodeId = int64(1)
for i := 1; i < len(ans); i++ {
(ans)[i].BeginNodeId = (ans)[i-1].GetEndNodeId()
}
}
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")
}
}
}()
// CreateDatabase creates a database if it doesn't exist
if err := pdb.Exec(fmt.Sprintf(createDatabaseQuery, name)); err != nil {
if pdb.IsDupDatabaseError(err) {
return nil
)
return string(Bytes([]byte(str)))
}
result := make([]byte, len(slice))
copy(result, slice)
rand.Shuffle(len(result), func(i, j int) {
result[i], result[j] = result[j], result[i]
})
return result
}
}
return ZapTag{
field: zap.String(key, value),
}
}
func NewStringsTag(key string, value []string) ZapTag {
}
return ZapTag{
field: zap.Bool(key, value),
}
}
func NewErrorTag(key string, value error) ZapTag {
}
d := metricDefinition{
name: name,
description: "",
unit: "",
}
for _, opt := range opts {
opt.apply(&d)
}
return d
}
// NewDataBlob returns a new DataBlob.
// TODO: return an UnknowEncodingType error with the actual type string when encodingTypeStr is invalid
encodingType, err := enumspb.EncodingTypeFromString(encodingTypeStr)
if err != nil {
encodingType = enumspb.ENCODING_TYPE_UNSPECIFIED
}
Data: data,
EncodingType: encodingType,
}
}
)
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 {
}
return ProtoAssertions{t}
}
// ProtoEqual compares two proto messages for equality using proto semantics. Options can be passed to customize
// generic parameters and slice casting (say from []historyEvent) to
// []proto.Message is impossible
if th, ok := t.(helper); ok {
th.Helper()
}
if len(a) != len(b) {
require.Fail(t, fmt.Sprintf("Proto slice length mismatch: want %d, got %d", len(a), len(b)))
return
}
require.Fail(t, fmt.Sprintf("Proto mismatch at index %d (-want +got):\n%v", i, diff))
return
)
return Category{
id: id,
cType: cType,
name: name,
}
}
func (c Category) ID() int {
}
return c.name
}
func (c Category) Type() CategoryType {
logger log.Logger,
serializer serialization.Serializer,
return &sqlExecutionStore{
SqlStore: NewSQLStore(db, logger, serializer),
HistoryBranchUtil: p.NewHistoryBranchUtil(serializer),
}, nil
}
// txExecuteShardLocked executes f under transaction and with read lock on shard row
}
return m.HistoryBranchUtil
}
func getStartTimeFromState(state *persistencespb.WorkflowExecutionState) *time.Time {
)
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:
// - error if input is malformed
// - UUID object if input can be parsed and is valid
if s == "" {
return nil, nil
}
if err != nil {
return nil, err
}
}
// Value implements sql.Valuer so that UUIDs can be written to databases
// transparently. This method returns a byte slice representation of uuid
return []byte(u), nil
}
func (u UUID) encodeHex(dst []byte) {
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")
}
}
)
// WithTags creates a new MetricProvder with provided []Tag
// Tags are merged with registered Tags from the source MetricsHandler
return n
}
// Counter obtains a counter for the given name.
return NoopCounterMetricFunc
}
// Gauge obtains a gauge for the given name.
}
var NoopGaugeMetricFunc = GaugeFunc(func(f float64, t ...Tag) {})
var NoopTimerMetricFunc = TimerFunc(func(d time.Duration, t ...Tag) {})
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) {
)
// We need this option to prevent faker.FakeData from returning an error for any struct that has an interface{} field.
faker.SetIgnoreInterface(true)
// We need this option to prevent faker from taking a long time while generating random data for structs that have
// map or slice fields. This is especially relevant for persistence.ShardInfo, which takes about 1s without this
// option, but only ~100µs with it.
if err := faker.SetRandomMapAndSliceMaxSize(2); err != nil {
panic(err)
}
// NewTransition creates a new [Transition] from the given source states to a destination state for a given event.
// The apply function is called after verifying the transition is possible and setting the destination state.
func NewTransition[S comparable, SM StateMachine[S], E any](src []S, dst S, apply func(SM, E) (TransitionOutput, error)) Transition[S, SM, E] {
sm.go
return Transition[S, SM, E]{
Sources: src,
Destination: dst,
apply: apply,
}
}
// Possible returns a boolean indicating whether the transition is possible for the current state.
)
var defaultProvider TaskGeneratorProvider = new(taskGeneratorProviderImpl)
populateTaskGeneratorProvider(defaultProvider)
}
func populateTaskGeneratorProvider(provider TaskGeneratorProvider) {
task_generator_provider.go
_taskGeneratorProvider.Store(&provider)
}
func GetTaskGeneratorProvider() TaskGeneratorProvider {
// 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
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)
}
// 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
)
switch k {
return "main"
case DbKindVisibility:
return "visibility"
return "unknown"
}
}
}
sql.RegisterPlugin(PluginName, &plugin{
queryConverter: &queryConverter{},
connPool: newConnPool(),
})
}
func (p *plugin) GetVisibilityQueryConverter() sqlplugin.VisibilityQueryConverter {
)
return &NoopResolver{}
}
return []string{service}
}
func ConvertWeightsToDynamicConfigValue(
weights map[tasks.Priority]int,
weightsForDC := make(map[string]any)
for priority, weight := range weights {
weightsForDC[priority.String()] = weight
}
return weightsForDC
}
}
return Key{
FireTime: fireTime,
TaskID: taskID,
}
}
func ValidateKey(key Key) error {
}
return backoff.NewExponentialRetryPolicy(cfg.InitialInterval).
WithMaximumInterval(cfg.MaxInterval).
WithExpirationInterval(backoff.NoInterval)
}
var defaultRetryPolicyConfig = RetryPolicyConfig{
// serializeConflictToken serializes a conflict token as a byte slice.
token := make([]byte, 8)
binary.LittleEndian.PutUint64(token, uint64(conflictToken))
return token
}
// newTaggedLogger returns a logger tagged with the Scheduler's attributes.
// 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
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
// ResolvePassword returns the database password, either from the static Password
// field or by executing PasswordCommand. If neither is set, it returns an empty string.
if c.PasswordCommand == nil {
}
timeout := c.PasswordCommand.Timeout
if timeout == 0 {
}
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()
}
}
// InverseMap creates the inverse map, ie., for a key-value map, it builds the value-key map.
if m == nil {
return nil
}
invm := make(map[V]K, len(m))
for k, v := range m {
// Size returns the size of the object, in bytes, once serialized
return proto.Size(val)
}
// Equal returns whether two Predicate values are equivalent by recursively
// tasks within the CHASM framework.
// The format of the returned FQN is: "libName.name"
return libName + "." + name
}
// The generated ID is used to uniquely identify components and tasks within the CHASM framework. The same FQN will
// always produce the same ID.
return farm.Fingerprint32([]byte(fqn))
}
// hasBusinessIDAlias returns true if the component has a businessID alias configured
)
return Key{handle: unique.Make(strings.ToLower(s))}
}
func (k Key) String() string {
// NewNoopLogger return a noopLogger
return &noopLogger{}
}
func (n *noopLogger) Debug(string, ...tag.Tag) {}
// NewHostInfoFromAddress creates a new HostInfo instance from a socket address.
return hostAddress(address)
}
// hostAddress is a HostInfo implementation that uses a string as the address and identity.
}
return Tag{Key: PersistenceDBKindTagName, Value: kind}
}
func HeaderCallsiteTag(kind string) Tag {
}
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()
}
// NewUnprocessableTaskError returns a new UnprocessableTaskError from given message.
return &UnprocessableTaskError{Message: message}
}
func (e UnprocessableTaskError) Error() string {
}
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 (c GaugeFunc) Record(v float64, tags ...Tag) { c(v, tags...) }
func (c TimerFunc) Record(v time.Duration, tags ...Tag) { c(v, tags...) }
)
func newNoopSignalAggregator() *noopSignalAggregator { return &noopSignalAggregator{} }
noop_health_signal_aggregator.go
func (a *noopSignalAggregator) Start() {}