Atlas › Test
Exact test identity: go.temporal.io/server/common/persistence/tests/TestSQLiteHistoryV2PersistenceSuite/TestConcurrentlyForkAndAppendBranches
go.temporal.io/server/common/persistence/testsTestSQLiteHistoryV2PersistenceSuite/TestConcurrentlyForkAndAppendBranchesTestConcurrentlyForkAndAppendBranchesExpand 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
history_manager.go
for _, br := range forkBranch.Ancestors {
if br.GetEndNodeId() >= request.ForkNodeID {
newAncestors = append(newAncestors, &persistencespb.HistoryBranchRange{
BranchId: br.GetBranchId(),
BeginNodeId: br.GetBeginNodeId(),
EndNodeId: request.ForkNodeID,
})
break
} else {
newAncestors = append(newAncestors, br)
}
}
// 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 *DeleteHistoryBranchRequest,
branch, err := m.GetHistoryBranchUtil().ParseHistoryBranchInfo(request.BranchToken)
if err != nil {
return serviceerror.NewInvalidArgument(fmt.Sprintf("unable to parse branch token: %v", err))
}
// However, it is possible that part of the target branch (or its ancestors) is used as ancestors by other branch.
// We need to avoid deleting those referenced parts. This is similar to reference count in garbage collection.
brsToDelete = append(brsToDelete, &persistencespb.HistoryBranchRange{
BranchId: branch.GetBranchId(),
BeginNodeId: GetBeginNodeID(branch),
})
// Get the history tree containing the branch to be delelted,
// so we know if any part of the target branch is referenced by other branches.
historyTreeResp, err := m.persistence.GetHistoryTreeContainingBranch(ctx, &InternalGetHistoryTreeContainingBranchRequest{
BranchToken: request.BranchToken,
ShardID: request.ShardID,
})
if err != nil {
return err
}
if err != nil {
return err
}
// usedBranches record branches referenced by others
for _, branchInfo := range branchInfos {
if branchInfo.BranchId == branch.BranchId {
// skip the target branch
continue
}
for _, ancestor := range branchInfo.Ancestors {
if curr, ok := usedBranches[ancestor.GetBranchId()]; !ok || curr < ancestor.GetEndNodeId() {
history_manager.go
usedBranches[ancestor.GetBranchId()] = ancestor.GetEndNodeId()
}
}
}
// for each branch range to delete, we iterate from bottom up, and stop when the range is also used by others
findDeleteRanges:
for i := len(brsToDelete) - 1; i >= 0; i-- {
br := brsToDelete[i]
if maxEndNode, ok := usedBranches[br.GetBranchId()]; ok {
if maxEndNode != common.LastEventID {
BranchId: br.BranchId,
BeginNodeId: maxEndNode,
})
}
// all ancestors are also used, no need to go up further,
// No other branch is using this range, we can delete all of it
deleteRanges = append(deleteRanges, InternalDeleteHistoryBranchRange{
BranchId: br.BranchId,
BeginNodeId: br.BeginNodeId,
})
}
}
BranchToken: request.BranchToken,
BranchInfo: branch,
ShardID: request.ShardID,
BranchRanges: deleteRanges,
}
return m.persistence.DeleteHistoryBranch(ctx, req)
}
func (m *executionManagerImpl) deserializeBranchInfos(
historyTreeResp *InternalGetHistoryTreeContainingBranchResponse,
branchInfos := make([]*persistencespb.HistoryBranch, 0, len(historyTreeResp.TreeInfos))
for _, blob := range historyTreeResp.TreeInfos {
treeInfo, err := m.serializer.HistoryTreeInfoFromBlob(blob)
if err != nil {
return nil, 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
}
}
Msg: "cannot append to ancestors' nodes",
}
}
}
ctx context.Context,
request *AppendHistoryNodesRequest,
req, err := m.serializeAppendHistoryNodesRequest(request)
if err != nil {
}
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
ctx context.Context,
request *GetAllHistoryTreeBranchesRequest,
resp, err := m.persistence.GetAllHistoryTreeBranches(ctx, request)
if err != nil {
return nil, err
}
for _, branch := range resp.Branches {
treeInfo, err := m.serializer.HistoryTreeInfoFromBlob(NewDataBlob(branch.Data, branch.Encoding))
history_manager.go
if err != nil {
return nil, err
}
BranchInfo: treeInfo.BranchInfo,
ForkTime: treeInfo.ForkTime,
Info: treeInfo.Info,
}
branches = append(branches, branchDetail)
}
NextPageToken: resp.NextPageToken,
Branches: branches,
}, nil
}
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 {
}
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)
}
}
)
switch err.(type) {
case *p.ConditionFailedError:
return true
return false
}
}
// SetupSuite implementation
}
// TearDownSuite implementation
s.TearDownWorkflowStore()
}
// SetupTest implementation
// Have to define our overridden assertions in the test setup. If we did it earlier, s.T() will return nil
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)
}
// TearDownTest implementation
s.cancel()
}
// TestGenUUIDs testing uuid.NewString() can generate unique UUID
// TestConcurrentlyForkAndAppendBranches test
func (s *HistoryV2PersistenceSuite) TestConcurrentlyForkAndAppendBranches() {
history_v2_persistence.go
treeID := uuid.NewString()
wg := sync.WaitGroup{}
concurrency := 10
masterBr, err := s.newHistoryBranch(treeID)
s.Nil(err)
branches := s.descTree(treeID)
s.Equal(0, len(branches))
// append first batch to master branch
eids := []int64{}
for i := int64(1); i <= int64(concurrency)+1; i++ {
eids = append(eids, i)
}
events := s.genRandomEvents(eids, 1)
err = s.appendNewBranchAndFirstNode(masterBr, events[0:1], 1, "masterbr")
s.Nil(err)
readEvents := s.read(masterBr, 1, int64(concurrency)+2)
s.Nil(err)
s.Equal(1, len(readEvents))
branches = s.descTree(treeID)
s.Equal(1, len(branches))
mbrID := branches[0].BranchId
txn := int64(1)
getTxnLock := sync.Mutex{}
reserveTxn := func(count int) int64 {
getTxnLock.Lock()
defer getTxnLock.Unlock()
ret := txn
txn += int64(count)
return ret
}
err = s.appendOneByOne(masterBr, events[1:], reserveTxn(len(events[1:])))
history_v2_persistence.go
s.Nil(err)
events = s.read(masterBr, 1, int64(concurrency)+2)
s.Nil(err)
s.Equal((concurrency)+1, len(events))
level1ID := new(sync.Map)
level1Br := new(sync.Map)
// test forking from master branch and append nodes
for i := range concurrency {
wg.Add(1)
go func(idx int) {
defer wg.Done()
forkNodeID := rand.Int63n(int64(concurrency)) + 2
level1ID.Store(idx, forkNodeID)
bi, err := s.fork(masterBr, forkNodeID)
s.Nil(err)
level1Br.Store(idx, bi)
// cannot append to ancestors
events := s.genRandomEvents([]int64{forkNodeID - 1}, 1)
err = s.appendNewNode(bi, events, reserveTxn(1))
_, ok := err.(*p.InvalidPersistenceRequestError)
s.Equal(true, ok)
// append second batch to first level
eids := make([]int64, 0)
for i := forkNodeID; i <= int64(concurrency)*2+1; i++ {
eids = append(eids, i)
}
events = s.genRandomEvents(eids, 1)
err = s.appendNewNode(bi, events[0:1], reserveTxn(1))
s.Nil(err)
err = s.appendOneByOne(bi, events[1:], reserveTxn(len(events[1:])))
s.Nil(err)
events = s.read(bi, 1, int64(concurrency)*2+2)
s.Nil(err)
s.Equal((concurrency)*2+1, len(events))
if idx == 0 {
err = s.deleteHistoryBranch(bi)
s.Nil(err)
}
}(i)
}
branches = s.descTree(treeID)
s.Equal(concurrency, len(branches))
forkOnLevel1 := int32(0)
level2Br := new(sync.Map)
wg = sync.WaitGroup{}
// test forking for second level of branch
for i := 1; i < concurrency; i++ {
wg.Add(1)
go func(idx int) {
defer wg.Done()
// Event we fork from level1 branch, it is possible that the new branch will fork from master branch
forkNodeID := rand.Int63n(int64(concurrency)*2) + 2
forkBr := s.getBranchByKey(level1Br, idx)
lastForkNodeID := s.getIDByKey(level1ID, idx)
if forkNodeID > lastForkNodeID {
atomic.AddInt32(&forkOnLevel1, int32(1))
}
s.Nil(err)
level2Br.Store(idx, bi)
// append second batch to second level
eids := make([]int64, 0)
for i := forkNodeID; i <= int64(concurrency)*3+1; i++ {
eids = append(eids, i)
}
events := s.genRandomEvents(eids, 1)
err = s.appendNewNode(bi, events[0:1], reserveTxn(1))
s.Nil(err)
err = s.appendOneByOne(bi, events[1:], reserveTxn(len(events[1:])))
s.Nil(err)
events = s.read(bi, 1, int64(concurrency)*3+2)
s.Nil(err)
s.Equal((concurrency)*3+1, len(events))
// try override last event
events = s.genRandomEvents([]int64{int64(concurrency)*3 + 1}, 1)
err = s.appendNewNode(bi, events, reserveTxn(1))
s.Nil(err)
events = s.read(bi, 1, int64(concurrency)*3+2)
s.Nil(err)
s.Equal((concurrency)*3+1, len(events))
// test fork and newBranch concurrently
bi, err = s.newHistoryBranch(treeID)
s.Nil(err)
level2Br.Store(concurrency+idx, bi)
events = s.genRandomEvents([]int64{1}, 1)
err = s.appendNewBranchAndFirstNode(bi, events, reserveTxn(1), "newbr")
s.Nil(err)
}(i)
}
branches = s.descTree(treeID)
s.Equal(concurrency*3-2, len(branches))
actualForkOnLevel1 := int32(0)
masterCnt := 0
for _, b := range branches {
if len(b.Ancestors) == 2 {
actualForkOnLevel1++
} else if len(b.Ancestors) == 0 {
masterCnt++
} else {
s.Equal(1, len(b.Ancestors))
s.Equal(mbrID, b.Ancestors[0].GetBranchId())
}
}
s.Equal(concurrency, masterCnt)
// Finally lets clean up all branches
level1Br.Range(func(k, v any) bool {
br := v.([]byte)
// delete old branches along with create new branches
err := s.deleteHistoryBranch(br)
s.Nil(err)
return true
})
level2Br.Range(func(k, v any) bool {
br := v.([]byte)
// delete old branches along with create new branches
err := s.deleteHistoryBranch(br)
s.Nil(err)
return true
})
err = s.deleteHistoryBranch(masterBr)
s.Nil(err)
branches = s.descTree(treeID)
s.Equal(0, len(branches))
}
func (s *HistoryV2PersistenceSuite) getBranchByKey(m *sync.Map, k int) []byte {
history_v2_persistence.go
v, ok := m.Load(k)
s.Equal(true, ok)
br := v.([]byte)
return br
}
func (s *HistoryV2PersistenceSuite) getIDByKey(m *sync.Map, k int) int64 {
history_v2_persistence.go
v, ok := m.Load(k)
s.Equal(true, ok)
id := v.(int64)
return id
}
func (s *HistoryV2PersistenceSuite) genRandomEvents(eventIDs []int64, version int64) []*historypb.HistoryEvent {
history_v2_persistence.go
var events []*historypb.HistoryEvent
now := time.Date(2020, 8, 22, 0, 0, 0, 0, time.UTC)
for _, eid := range eventIDs {
e := &historypb.HistoryEvent{EventId: eid, Version: version, EventTime: timestamppb.New(now)}
events = append(events, e)
}
}
// persistence helper
func (s *HistoryV2PersistenceSuite) newHistoryBranch(treeID string) ([]byte, error) {
history_v2_persistence.go
return s.ExecutionManager.GetHistoryBranchUtil().NewHistoryBranch(
uuid.NewString(),
uuid.NewString(),
uuid.NewString(),
treeID,
nil,
[]*persistencespb.HistoryBranchRange{},
0,
0,
0,
)
}
// persistence helper
func (s *HistoryV2PersistenceSuite) deleteHistoryBranch(branch []byte) error {
history_v2_persistence.go
op := func() error {
return s.ExecutionManager.DeleteHistoryBranch(s.ctx, &p.DeleteHistoryBranchRequest{
BranchToken: branch,
ShardID: s.ShardInfo.GetShardId(),
})
}
return backoff.ThrottleRetry(op, historyTestRetryPolicy, isConditionFail)
history_v2_persistence.go
}
// persistence helper
func (s *HistoryV2PersistenceSuite) descTree(treeID string) []*persistencespb.HistoryBranch {
history_v2_persistence.go
var branches []*persistencespb.HistoryBranch
var nextPageToken []byte
for {
resp, err := s.ExecutionManager.GetAllHistoryTreeBranches(s.ctx, &p.GetAllHistoryTreeBranchesRequest{
NextPageToken: nextPageToken,
PageSize: 100,
})
s.NoError(err)
for _, branch := range resp.Branches {
branches = append(branches, branch.BranchInfo)
}
}
if len(nextPageToken) == 0 {
break
}
}
}
// persistence helper
func (s *HistoryV2PersistenceSuite) read(branch []byte, minID, maxID int64) []*historypb.HistoryEvent {
history_v2_persistence.go
res, err := s.readWithError(branch, minID, maxID)
s.Nil(err)
return res
}
func (s *HistoryV2PersistenceSuite) readWithError(branch []byte, minID, maxID int64) ([]*historypb.HistoryEvent, error) {
history_v2_persistence.go
// use small page size to enforce pagination
randPageSize := 2
res := make([]*historypb.HistoryEvent, 0)
token := []byte{}
for {
resp, err := s.ExecutionManager.ReadHistoryBranch(s.ctx, &p.ReadHistoryBranchRequest{
BranchToken: branch,
MinEventID: minID,
MaxEventID: maxID,
PageSize: randPageSize,
NextPageToken: token,
ShardID: s.ShardInfo.GetShardId(),
})
if err != nil {
return nil, err
}
s.True(resp.Size > 0)
}
res = append(res, resp.HistoryEvents...)
token = resp.NextPageToken
if len(token) == 0 {
break
}
}
}
func (s *HistoryV2PersistenceSuite) appendOneByOne(branch []byte, events []*historypb.HistoryEvent, txnID int64) error {
history_v2_persistence.go
for index, e := range events {
err := s.append(branch, []*historypb.HistoryEvent{e}, txnID+int64(index), false, "")
if err != nil {
return err
}
}
}
func (s *HistoryV2PersistenceSuite) appendNewNode(branch []byte, events []*historypb.HistoryEvent, txnID int64) error {
history_v2_persistence.go
return s.append(branch, events, txnID, false, "")
}
func (s *HistoryV2PersistenceSuite) appendNewBranchAndFirstNode(branch []byte, events []*historypb.HistoryEvent, txnID int64, branchInfo string) error {
history_v2_persistence.go
return s.append(branch, events, txnID, true, branchInfo)
}
// persistence helper
func (s *HistoryV2PersistenceSuite) append(branch []byte, events []*historypb.HistoryEvent, txnID int64, isNewBranch bool, branchInfo string) error {
history_v2_persistence.go
var resp *p.AppendHistoryNodesResponse
op := func() error {
var err error
resp, err = s.ExecutionManager.AppendHistoryNodes(s.ctx, &p.AppendHistoryNodesRequest{
IsNewBranch: isNewBranch,
Info: branchInfo,
BranchToken: branch,
Events: events,
TransactionID: txnID,
ShardID: s.ShardInfo.GetShardId(),
})
return err
}
err := backoff.ThrottleRetry(op, historyTestRetryPolicy, isConditionFail)
history_v2_persistence.go
if err != nil {
}
return err
}
// persistence helper
func (s *HistoryV2PersistenceSuite) fork(forkBranch []byte, forkNodeID int64) ([]byte, error) {
history_v2_persistence.go
bi := []byte{}
op := func() error {
var err error
resp, err := s.ExecutionManager.ForkHistoryBranch(s.ctx, &p.ForkHistoryBranchRequest{
ForkBranchToken: forkBranch,
ForkNodeID: forkNodeID,
Info: testForkRunID,
ShardID: s.ShardInfo.GetShardId(),
NamespaceID: uuid.NewString(),
NewRunID: uuid.NewString(),
})
if resp != nil {
bi = resp.NewBranchToken
}
return err
}
err := backoff.ThrottleRetry(op, historyTestRetryPolicy, isConditionFail)
history_v2_persistence.go
return bi, err
}
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 {
type TypedPropertyFn[T any] func() T
return func() T {
prec := []Constraints{{}}
return matchAndConvert(
}
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 {
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.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)
}
}
ctx context.Context,
request *p.InternalDeleteHistoryBranchRequest,
branchIDBytes, err := primitives.ParseUUID(request.BranchInfo.BranchId)
if err != nil {
return err
}
if err != nil {
return err
}
_, err = tx.DeleteFromHistoryTree(ctx, sqlplugin.HistoryTreeDeleteFilter{
TreeID: treeIDBytes,
BranchID: branchIDBytes,
ShardID: request.ShardID,
})
if err != nil {
return err
}
// delete each branch range
branchIDBytes, err := primitives.ParseUUID(br.BranchId)
if err != nil {
return err
}
ShardID: request.ShardID,
TreeID: treeIDBytes,
BranchID: branchIDBytes,
MinNodeID: br.BeginNodeId,
}
_, err = tx.RangeDeleteFromHistoryNode(ctx, deleteFilter)
if err != nil {
return err
}
}
})
}
ctx context.Context,
request *p.GetAllHistoryTreeBranchesRequest,
pageSize := request.PageSize
if pageSize <= 0 {
return nil, fmt.Errorf("PageSize must be greater than 0, but was %d", pageSize)
}
Limit: pageSize,
}
if len(request.NextPageToken) != 0 {
var token getAllHistoryTreeBranchesPaginationToken
if err := json.Unmarshal(request.NextPageToken, &token); err != nil {
}
if err != nil {
return nil, err
}
for _, row := range rows {
TreeID: row.TreeID.String(),
BranchID: row.BranchID.String(),
Data: row.Data,
Encoding: row.DataEncoding,
}
branches = append(branches, branch)
}
Branches: branches,
}
if len(branches) < pageSize {
// no next page token because there are no more results
return response, nil
}
// if we filled the page with rows, then set the next page token
ctx context.Context,
request *p.InternalGetHistoryTreeContainingBranchRequest,
branch, err := m.ParseHistoryBranchInfo(request.BranchToken)
if err != nil {
return nil, err
}
if err != nil {
return nil, err
}
rows, err := m.DB.SelectFromHistoryTree(ctx, sqlplugin.HistoryTreeSelectFilter{
history_store.go
TreeID: treeID,
ShardID: request.ShardID,
})
if err == sql.ErrNoRows || (err == nil && len(rows) == 0) {
return &p.InternalGetHistoryTreeContainingBranchResponse{}, nil
}
for _, row := range rows {
treeInfos = append(treeInfos, p.NewDataBlob(row.Data, row.DataEncoding))
}
TreeInfos: treeInfos,
}, nil
}
// NewShardPersistenceMetricsClient creates a client to manage shards
func NewShardPersistenceMetricsClient(persistence ShardManager, metricsHandler metrics.Handler, healthSignals HealthSignalAggregator, logger log.Logger, enableDataLossMetrics dynamicconfig.BoolPropertyFn) ShardManager {
persistence_metric_clients.go
return &shardPersistenceClient{
metricEmitter: metricEmitter{
metricsHandler: metricsHandler,
logger: logger,
enableDataLossMetrics: enableDataLossMetrics,
},
healthSignals: healthSignals,
persistence: persistence,
}
}
// NewExecutionPersistenceMetricsClient creates a client to manage executions
func NewExecutionPersistenceMetricsClient(persistence ExecutionManager, metricsHandler metrics.Handler, healthSignals HealthSignalAggregator, logger log.Logger, enableDataLossMetrics dynamicconfig.BoolPropertyFn) ExecutionManager {
persistence_metric_clients.go
return &executionPersistenceClient{
metricEmitter: metricEmitter{
metricsHandler: metricsHandler,
logger: logger,
enableDataLossMetrics: enableDataLossMetrics,
},
healthSignals: healthSignals,
persistence: persistence,
}
}
// NewTaskPersistenceMetricsClient creates a client to manage tasks
func NewTaskPersistenceMetricsClient(persistence TaskManager, metricsHandler metrics.Handler, healthSignals HealthSignalAggregator, logger log.Logger, enableDataLossMetrics dynamicconfig.BoolPropertyFn) TaskManager {
persistence_metric_clients.go
return &taskPersistenceClient{
metricEmitter: metricEmitter{
metricsHandler: metricsHandler,
logger: logger,
enableDataLossMetrics: enableDataLossMetrics,
},
healthSignals: healthSignals,
persistence: persistence,
}
}
// NewMetadataPersistenceMetricsClient creates a MetadataManager client to manage metadata
func NewMetadataPersistenceMetricsClient(persistence MetadataManager, metricsHandler metrics.Handler, healthSignals HealthSignalAggregator, logger log.Logger, enableDataLossMetrics dynamicconfig.BoolPropertyFn) MetadataManager {
persistence_metric_clients.go
return &metadataPersistenceClient{
metricEmitter: metricEmitter{
metricsHandler: metricsHandler,
logger: logger,
enableDataLossMetrics: enableDataLossMetrics,
},
healthSignals: healthSignals,
persistence: persistence,
}
}
// NewClusterMetadataPersistenceMetricsClient creates a ClusterMetadataManager client to manage cluster metadata
func NewClusterMetadataPersistenceMetricsClient(persistence ClusterMetadataManager, metricsHandler metrics.Handler, healthSignals HealthSignalAggregator, logger log.Logger, enableDataLossMetrics dynamicconfig.BoolPropertyFn) ClusterMetadataManager {
persistence_metric_clients.go
return &clusterMetadataPersistenceClient{
metricEmitter: metricEmitter{
metricsHandler: metricsHandler,
logger: logger,
enableDataLossMetrics: enableDataLossMetrics,
},
healthSignals: healthSignals,
persistence: persistence,
}
}
// NewQueuePersistenceMetricsClient creates a client to manage queue
func NewQueuePersistenceMetricsClient(persistence Queue, metricsHandler metrics.Handler, healthSignals HealthSignalAggregator, logger log.Logger, enableDataLossMetrics dynamicconfig.BoolPropertyFn) Queue {
persistence_metric_clients.go
return &queuePersistenceClient{
metricEmitter: metricEmitter{
metricsHandler: metricsHandler,
logger: logger,
enableDataLossMetrics: enableDataLossMetrics,
},
healthSignals: healthSignals,
persistence: persistence,
}
}
// NewNexusEndpointPersistenceMetricsClient creates a NexusEndpointManager to manage nexus endpoints
func NewNexusEndpointPersistenceMetricsClient(persistence NexusEndpointManager, metricsHandler metrics.Handler, healthSignals HealthSignalAggregator, logger log.Logger, enableDataLossMetrics dynamicconfig.BoolPropertyFn) NexusEndpointManager {
persistence_metric_clients.go
return &nexusEndpointPersistenceClient{
metricEmitter: metricEmitter{
metricsHandler: metricsHandler,
logger: logger,
enableDataLossMetrics: enableDataLossMetrics,
},
healthSignals: healthSignals,
persistence: persistence,
}
}
func (p *shardPersistenceClient) GetName() string {
ctx context.Context,
request *GetOrCreateShardRequest,
caller := headers.GetCallerInfo(ctx).CallerName
startTime := time.Now().UTC()
defer func() {
latency := time.Since(startTime)
p.healthSignals.Record(request.ShardID, latency, retErr)
p.recordRequestMetrics(metrics.PersistenceGetOrCreateShardScope, caller, latency, retErr)
p.recordDataLossMetrics(metrics.PersistenceGetOrCreateShardScope, caller, retErr, "", "")
}()
return p.persistence.GetOrCreateShard(ctx, request)
}
}
p.persistence.Close()
}
func (p *executionPersistenceClient) GetName() string {
}
func (p *executionPersistenceClient) GetHistoryBranchUtil() HistoryBranchUtil {
persistence_metric_clients.go
return p.persistence.GetHistoryBranchUtil()
}
func (p *executionPersistenceClient) CreateWorkflowExecution(
}
p.persistence.Close()
}
func (p *taskPersistenceClient) GetName() string {
}
p.persistence.Close()
}
func (p *metadataPersistenceClient) GetName() string {
}
p.persistence.Close()
}
// AppendHistoryNodes add a node to history node table
ctx context.Context,
request *AppendHistoryNodesRequest,
caller := headers.GetCallerInfo(ctx).CallerName
startTime := time.Now().UTC()
defer func() {
p.healthSignals.Record(CallerSegmentMissing, time.Since(startTime), retErr)
p.recordRequestMetrics(metrics.PersistenceAppendHistoryNodesScope, caller, time.Since(startTime), retErr)
p.recordDataLossMetrics(metrics.PersistenceAppendHistoryNodesScope, caller, retErr, "", "")
}()
return p.persistence.AppendHistoryNodes(ctx, request)
}
ctx context.Context,
request *ReadHistoryBranchRequest,
caller := headers.GetCallerInfo(ctx).CallerName
startTime := time.Now().UTC()
defer func() {
p.recordRequestMetrics(metrics.PersistenceReadHistoryBranchScope, caller, time.Since(startTime), retErr)
p.recordDataLossMetrics(metrics.PersistenceReadHistoryBranchScope, caller, retErr, "", "")
}()
return p.persistence.ReadHistoryBranch(ctx, request)
}
ctx context.Context,
request *ForkHistoryBranchRequest,
caller := headers.GetCallerInfo(ctx).CallerName
startTime := time.Now().UTC()
defer func() {
p.recordRequestMetrics(metrics.PersistenceForkHistoryBranchScope, caller, time.Since(startTime), retErr)
p.recordDataLossMetrics(metrics.PersistenceForkHistoryBranchScope, caller, retErr, "", "")
}()
return p.persistence.ForkHistoryBranch(ctx, request)
}
ctx context.Context,
request *DeleteHistoryBranchRequest,
caller := headers.GetCallerInfo(ctx).CallerName
startTime := time.Now().UTC()
defer func() {
p.recordRequestMetrics(metrics.PersistenceDeleteHistoryBranchScope, caller, time.Since(startTime), retErr)
p.recordDataLossMetrics(metrics.PersistenceDeleteHistoryBranchScope, caller, retErr, "", "")
}()
return p.persistence.DeleteHistoryBranch(ctx, request)
}
ctx context.Context,
request *GetAllHistoryTreeBranchesRequest,
caller := headers.GetCallerInfo(ctx).CallerName
startTime := time.Now().UTC()
defer func() {
p.healthSignals.Record(CallerSegmentMissing, time.Since(startTime), retErr)
p.recordRequestMetrics(metrics.PersistenceGetAllHistoryTreeBranchesScope, caller, time.Since(startTime), retErr)
p.recordDataLossMetrics(metrics.PersistenceGetAllHistoryTreeBranchesScope, caller, retErr, "", "")
}()
return p.persistence.GetAllHistoryTreeBranches(ctx, request)
}
ctx context.Context,
blob *commonpb.DataBlob,
return p.persistence.Init(ctx, blob)
}
func (p *queuePersistenceClient) EnqueueMessage(
}
p.persistence.Close()
}
p.persistence.Close()
}
func (p *clusterMetadataPersistenceClient) ListClusterMetadata(
}
p.persistence.Close()
}
func (p *nexusEndpointPersistenceClient) GetNexusEndpoint(
}
func (p *metricEmitter) recordRequestMetrics(operation string, caller string, latency time.Duration, err error) {
persistence_metric_clients.go
handler := p.metricsHandler.WithTags(metrics.OperationTag(operation), metrics.NamespaceTag(caller))
metrics.PersistenceRequests.With(handler).Record(1)
metrics.PersistenceLatency.With(handler).Record(latency)
updateErrorMetric(handler, p.logger, operation, err)
}
func (p *metricEmitter) recordDataLossMetrics(operation string, caller string, err error, workflowID, runID string) {
persistence_metric_clients.go
// Emit data loss metrics if enabled and error is DataLoss
var dataLoss *serviceerror.DataLoss
if errors.As(err, &dataLoss) {
if p.enableDataLossMetrics() {
EmitDataLossMetric(p.metricsHandler, caller, workflowID, runID, operation, err)
}
func updateErrorMetric(handler metrics.Handler, logger log.Logger, operation string, err error) {
persistence_metric_clients.go
if err != nil {
metrics.PersistenceErrorWithType.With(handler).Record(1, metrics.ServiceErrorTypeTag(err))
persistence_metric_clients.go
if common.IsContextCanceledErr(err) {
// no-op
return
}
case *ShardAlreadyExistError,
*ShardOwnershipLostError,
metrics.PersistenceErrResourceExhaustedCounter.With(handler).Record(
1, metrics.ResourceExhaustedCauseTag(err.Cause), metrics.ResourceExhaustedScopeTag(err.Scope))
logger.Error("Operation failed with internal error.", tag.Error(err), tag.ErrorType(err), tag.Operation(operation))
metrics.PersistenceFailures.With(handler).Record(1)
}
}
policy backoff.RetryPolicy,
isRetryable backoff.IsRetryable,
return &shardRetryablePersistenceClient{
persistence: persistence,
policy: policy,
isRetryable: isRetryable,
}
}
// NewExecutionPersistenceRetryableClient creates a client to manage executions
policy backoff.RetryPolicy,
isRetryable backoff.IsRetryable,
return &executionRetryablePersistenceClient{
persistence: persistence,
policy: policy,
isRetryable: isRetryable,
}
}
// NewTaskPersistenceRetryableClient creates a client to manage tasks
policy backoff.RetryPolicy,
isRetryable backoff.IsRetryable,
return &taskRetryablePersistenceClient{
persistence: persistence,
policy: policy,
isRetryable: isRetryable,
}
}
// NewMetadataPersistenceRetryableClient creates a MetadataManager client to manage metadata
policy backoff.RetryPolicy,
isRetryable backoff.IsRetryable,
return &metadataRetryablePersistenceClient{
persistence: persistence,
policy: policy,
isRetryable: isRetryable,
}
}
// NewClusterMetadataPersistenceRetryableClient creates a ClusterMetadataManager client to manage cluster metadata
policy backoff.RetryPolicy,
isRetryable backoff.IsRetryable,
return &clusterMetadataRetryablePersistenceClient{
persistence: persistence,
policy: policy,
isRetryable: isRetryable,
}
}
// NewQueuePersistenceRetryableClient creates a client to manage queue
policy backoff.RetryPolicy,
isRetryable backoff.IsRetryable,
return &queueRetryablePersistenceClient{
persistence: persistence,
policy: policy,
isRetryable: isRetryable,
}
}
// NewNexusEndpointPersistenceRetryableClient creates a NexusEndpointManager client to manage nexus endpoints
policy backoff.RetryPolicy,
isRetryable backoff.IsRetryable,
return &nexusEndpointRetryablePersistenceClient{
persistence: persistence,
policy: policy,
isRetryable: isRetryable,
}
}
func (p *shardRetryablePersistenceClient) GetName() string {
ctx context.Context,
request *GetOrCreateShardRequest,
var response *GetOrCreateShardResponse
op := func(ctx context.Context) error {
var err error
response, err = p.persistence.GetOrCreateShard(ctx, request)
return err
}
err := backoff.ThrottleRetryContext(ctx, op, p.policy, p.isRetryable)
persistence_retryable_clients.go
return response, err
}
}
p.persistence.Close()
}
func (p *executionRetryablePersistenceClient) GetName() string {
}
func (p *executionRetryablePersistenceClient) GetHistoryBranchUtil() HistoryBranchUtil {
persistence_retryable_clients.go
return p.persistence.GetHistoryBranchUtil()
}
func (p *executionRetryablePersistenceClient) CreateWorkflowExecution(
ctx context.Context,
request *AppendHistoryNodesRequest,
var response *AppendHistoryNodesResponse
op := func(ctx context.Context) error {
var err error
response, err = p.persistence.AppendHistoryNodes(ctx, request)
return err
}
err := backoff.ThrottleRetryContext(ctx, op, p.policy, p.isRetryable)
persistence_retryable_clients.go
return response, err
}
ctx context.Context,
request *ReadHistoryBranchRequest,
var response *ReadHistoryBranchResponse
op := func(ctx context.Context) error {
var err error
response, err = p.persistence.ReadHistoryBranch(ctx, request)
return err
}
err := backoff.ThrottleRetryContext(ctx, op, p.policy, p.isRetryable)
persistence_retryable_clients.go
return response, err
}
ctx context.Context,
request *ForkHistoryBranchRequest,
var response *ForkHistoryBranchResponse
op := func(ctx context.Context) error {
var err error
response, err = p.persistence.ForkHistoryBranch(ctx, request)
return err
}
err := backoff.ThrottleRetryContext(ctx, op, p.policy, p.isRetryable)
persistence_retryable_clients.go
return response, err
}
ctx context.Context,
request *DeleteHistoryBranchRequest,
op := func(ctx context.Context) error {
return p.persistence.DeleteHistoryBranch(ctx, request)
}
return backoff.ThrottleRetryContext(ctx, op, p.policy, p.isRetryable)
persistence_retryable_clients.go
}
ctx context.Context,
request *GetAllHistoryTreeBranchesRequest,
var response *GetAllHistoryTreeBranchesResponse
op := func(ctx context.Context) error {
var err error
response, err = p.persistence.GetAllHistoryTreeBranches(ctx, request)
return err
}
err := backoff.ThrottleRetryContext(ctx, op, p.policy, p.isRetryable)
persistence_retryable_clients.go
return response, err
}
p.persistence.Close()
}
func (p *taskRetryablePersistenceClient) GetName() string {
}
p.persistence.Close()
}
func (p *metadataRetryablePersistenceClient) GetName() string {
}
p.persistence.Close()
}
func (p *clusterMetadataRetryablePersistenceClient) GetName() string {
}
func (p *clusterMetadataRetryablePersistenceClient) Close() {
persistence_retryable_clients.go
p.persistence.Close()
}
func (p *queueRetryablePersistenceClient) Init(
ctx context.Context,
blob *commonpb.DataBlob,
op := func(ctx context.Context) error {
return p.persistence.Init(ctx, blob)
}
return backoff.ThrottleRetryContext(ctx, op, p.policy, p.isRetryable)
persistence_retryable_clients.go
}
}
p.persistence.Close()
}
func (p *nexusEndpointRetryablePersistenceClient) GetName() string {
}
p.persistence.Close()
}
func (p *nexusEndpointRetryablePersistenceClient) GetNexusEndpoint(
// NewTestBaseWithSQL returns a new persistence test base backed by SQL
logger := options.Logger
if logger == nil {
}
case mysql.PluginName:
options.DBPort = environment.GetMySQLPort()
case postgresql.PluginName, postgresql.PluginNamePGX:
options.DBPort = environment.GetPostgreSQLPort()
options.DBPort = 0
default:
panic(fmt.Sprintf("unknown sql store driver: %v", options.SQLDBPluginName))
}
}
switch options.SQLDBPluginName {
case mysql.PluginName:
}
}
testCluster := sql.NewTestCluster(options.SQLDBPluginName, options.DBName, options.DBUsername, options.DBPassword, options.DBHost, options.DBPort, options.ConnectAttributes, options.SchemaDir, options.FaultInjection, logger)
persistence_test_base.go
return NewTestBaseForCluster(testCluster, logger)
}
}
func NewTestBaseForCluster(testCluster PersistenceTestCluster, logger log.Logger) *TestBase {
persistence_test_base.go
return &TestBase{
DefaultTestCluster: testCluster,
Logger: logger,
TracerProvider: telemetry.NoopTracerProvider,
}
}
// Setup sets up the test base, must be called as part of SetupSuite
var err error
shardID := int32(10)
if clusterMetadataConfig == nil {
clusterMetadataConfig = cluster.NewTestClusterMetadataConfig(false, false)
persistence_test_base.go
}
s.PersistenceHealthSignals = persistence.NoopHealthSignalAggregator
}
s.DefaultTestCluster.SetupTestDatabase()
cfg := s.DefaultTestCluster.Config()
serializer := serialization.NewSerializer()
dataStoreFactory := client.DataStoreFactoryProvider(
client.ClusterName(clusterName),
resolver.NewNoopResolver(),
&cfg,
s.AbstractDataStoreFactory,
s.Logger,
metrics.NoopMetricsHandler,
s.TracerProvider,
serializer,
)
factory := client.NewFactory(
dataStoreFactory,
&cfg,
s.PersistenceRateLimiter,
quotas.NoopRequestRateLimiter,
quotas.NoopRequestRateLimiter,
serializer,
nil,
clusterName,
metrics.NoopMetricsHandler,
s.Logger,
s.PersistenceHealthSignals,
func() bool { return false },
func() bool { return false },
)
s.fatalOnError("NewTaskManager", err)
s.FairTaskMgr, err = factory.NewFairTaskManager()
// TODO: re-enable error check after FairTaskManager is implemented for sql
// s.fatalOnError("NewFairTaskManager", err)
_ = err
s.ClusterMetadataManager, err = factory.NewClusterMetadataManager()
s.fatalOnError("NewClusterMetadataManager", err)
s.ClusterMetadata = cluster.NewMetadataFromConfig(clusterMetadataConfig, s.ClusterMetadataManager, dynamicconfig.NewNoopCollection(), s.Logger)
s.SearchAttributesManager = searchattribute.NewManager(
clock.NewRealTimeSource(),
s.ClusterMetadataManager,
s.Logger,
dynamicconfig.GetBoolPropertyFn(true),
)
s.MetadataManager, err = factory.NewMetadataManager()
s.fatalOnError("NewMetadataManager", err)
s.ShardMgr, err = factory.NewShardManager()
s.fatalOnError("NewShardManager", err)
s.ExecutionManager, err = factory.NewExecutionManager()
s.fatalOnError("NewExecutionManager", err)
s.NexusEndpointManager, err = factory.NewNexusEndpointManager()
s.fatalOnError("NewNexusEndpointManager", err)
s.Factory = factory
s.ReadLevel = 0
s.ReplicationReadLevel = 0
s.ShardInfo = &persistencespb.ShardInfo{
ShardId: shardID,
RangeId: 0,
}
s.TaskIDGenerator = &TestTransferTaskIDGenerator{}
_, err = s.ShardMgr.GetOrCreateShard(context.Background(), &persistence.GetOrCreateShardRequest{
ShardID: shardID,
InitialShardInfo: s.ShardInfo,
})
s.fatalOnError("CreateShard", err)
queue, err := factory.NewNamespaceReplicationQueue()
s.fatalOnError("Create NamespaceReplicationQueue", err)
s.NamespaceReplicationQueue = queue
}
if err != nil {
s.Logger.Fatal(msg, tag.Error(err))
}
// TearDownWorkflowStore to cleanup
s.TaskMgr.Close()
s.ClusterMetadataManager.Close()
s.MetadataManager.Close()
s.ExecutionManager.Close()
s.ShardMgr.Close()
s.ExecutionManager.Close()
s.NexusEndpointManager.Close()
s.NamespaceReplicationQueue.Close()
s.Factory.Close()
s.DefaultTestCluster.TearDownTestDatabase()
}
// EqualTimesWithPrecision assertion that two times are equal within precision
}
uuidPart := strings.ReplaceAll(uuid.NewString(), "-", "")
// Keep generated DB names short enough for Cassandra keyspaces after XDC tests append cluster suffixes.
return "test_" + uuidPart[:24]
}
func timeComparator(t1, t2 time.Time, timeTolerance time.Duration) bool {
enableDataLossMetrics EnableDataLossMetrics,
enableBestEffortDeleteTasksOnWorkflowUpdate EnableBestEffortDeleteTasksOnWorkflowUpdate,
factory := &factoryImpl{
dataStoreFactory: dataStoreFactory,
config: cfg,
serializer: serializer,
eventBlobCache: eventBlobCache,
metricsHandler: metricsHandler,
logger: logger,
clusterName: clusterName,
systemRateLimiter: systemRateLimiter,
namespaceRateLimiter: namespaceRateLimiter,
shardRateLimiter: shardRateLimiter,
healthSignals: healthSignals,
enableDataLossMetrics: dynamicconfig.BoolPropertyFn(enableDataLossMetrics),
enableBestEffortDeleteTasksOnWorkflowUpdate: dynamicconfig.BoolPropertyFn(enableBestEffortDeleteTasksOnWorkflowUpdate),
}
factory.initDependencies()
return factory
}
// NewTaskManager returns a new task manager
taskStore, err := f.dataStoreFactory.NewTaskStore()
if err != nil {
return nil, err
}
if f.systemRateLimiter != nil && f.namespaceRateLimiter != nil {
result = persistence.NewTaskPersistenceRateLimitedClient(result, f.systemRateLimiter, f.namespaceRateLimiter, f.shardRateLimiter, f.logger)
}
result = persistence.NewTaskPersistenceMetricsClient(result, f.metricsHandler, f.healthSignals, f.logger, f.enableDataLossMetrics)
}
result = persistence.NewTaskPersistenceRetryableClient(result, retryPolicy, IsPersistenceTransientError)
return result, nil
}
// NewFairTaskManager returns a new task fairness manager
taskStore, err := f.dataStoreFactory.NewFairTaskStore()
if err != nil {
return nil, err
}
if f.systemRateLimiter != nil && f.namespaceRateLimiter != nil {
result = persistence.NewTaskPersistenceRateLimitedClient(result, f.systemRateLimiter, f.namespaceRateLimiter, f.shardRateLimiter, f.logger)
}
result = persistence.NewTaskPersistenceMetricsClient(result, f.metricsHandler, f.healthSignals, f.logger, f.enableDataLossMetrics)
}
result = persistence.NewTaskPersistenceRetryableClient(result, retryPolicy, IsPersistenceTransientError)
return result, nil
}
// NewShardManager returns a new shard manager
shardStore, err := f.dataStoreFactory.NewShardStore()
if err != nil {
return nil, err
}
if f.systemRateLimiter != nil && f.namespaceRateLimiter != nil {
result = persistence.NewShardPersistenceRateLimitedClient(result, f.systemRateLimiter, f.namespaceRateLimiter, f.shardRateLimiter, f.logger)
}
result = persistence.NewShardPersistenceMetricsClient(result, f.metricsHandler, f.healthSignals, f.logger, f.enableDataLossMetrics)
}
result = persistence.NewShardPersistenceRetryableClient(result, retryPolicy, IsPersistenceTransientError)
return result, nil
}
// NewMetadataManager returns a new metadata manager
store, err := f.dataStoreFactory.NewMetadataStore()
if err != nil {
return nil, err
}
result := persistence.NewMetadataManagerImpl(store, f.serializer, f.logger, f.clusterName)
factory.go
if f.systemRateLimiter != nil && f.namespaceRateLimiter != nil {
result = persistence.NewMetadataPersistenceRateLimitedClient(result, f.systemRateLimiter, f.namespaceRateLimiter, f.shardRateLimiter, f.logger)
}
result = persistence.NewMetadataPersistenceMetricsClient(result, f.metricsHandler, f.healthSignals, f.logger, f.enableDataLossMetrics)
}
result = persistence.NewMetadataPersistenceRetryableClient(result, retryPolicy, IsPersistenceTransientError)
return result, nil
}
// NewClusterMetadataManager returns a new cluster metadata manager
func (f *factoryImpl) NewClusterMetadataManager() (persistence.ClusterMetadataManager, error) {
factory.go
store, err := f.dataStoreFactory.NewClusterMetadataStore()
if err != nil {
return nil, err
}
result := persistence.NewClusterMetadataManagerImpl(store, f.serializer, f.clusterName, f.logger)
factory.go
if f.systemRateLimiter != nil && f.namespaceRateLimiter != nil {
result = persistence.NewClusterMetadataPersistenceRateLimitedClient(result, f.systemRateLimiter, f.namespaceRateLimiter, f.shardRateLimiter, f.logger)
}
result = persistence.NewClusterMetadataPersistenceMetricsClient(result, f.metricsHandler, f.healthSignals, f.logger, f.enableDataLossMetrics)
}
result = persistence.NewClusterMetadataPersistenceRetryableClient(result, retryPolicy, IsPersistenceTransientError)
return result, nil
}
// NewExecutionManager returns a new execution manager
func (f *factoryImpl) NewExecutionManager() (persistence.ExecutionManager, error) {
factory.go
store, err := f.dataStoreFactory.NewExecutionStore()
if err != nil {
return nil, err
}
store,
f.serializer,
f.eventBlobCache,
f.logger,
f.config.TransactionSizeLimit,
f.enableBestEffortDeleteTasksOnWorkflowUpdate,
)
if f.systemRateLimiter != nil && f.namespaceRateLimiter != nil {
result = persistence.NewExecutionPersistenceRateLimitedClient(result, f.systemRateLimiter, f.namespaceRateLimiter, f.shardRateLimiter, f.logger)
}
result = persistence.NewExecutionPersistenceMetricsClient(result, f.metricsHandler, f.healthSignals, f.logger, f.enableDataLossMetrics)
}
result = persistence.NewExecutionPersistenceRetryableClient(result, retryPolicy, IsPersistenceTransientError)
return result, nil
}
func (f *factoryImpl) NewNamespaceReplicationQueue() (persistence.NamespaceReplicationQueue, error) {
factory.go
result, err := f.dataStoreFactory.NewQueue(persistence.NamespaceReplicationQueueType)
if err != nil {
return nil, err
}
result = persistence.NewQueuePersistenceRateLimitedClient(result, f.systemRateLimiter, f.namespaceRateLimiter, f.shardRateLimiter, f.logger)
}
result = persistence.NewQueuePersistenceMetricsClient(result, f.metricsHandler, f.healthSignals, f.logger, f.enableDataLossMetrics)
}
result = persistence.NewQueuePersistenceRetryableClient(result, namespaceQueueRetryPolicy, IsNamespaceQueueTransientError)
return persistence.NewNamespaceReplicationQueue(result, f.serializer, f.clusterName, f.metricsHandler, f.logger)
}
}
func (f *factoryImpl) NewNexusEndpointManager() (persistence.NexusEndpointManager, error) {
factory.go
store, err := f.dataStoreFactory.NewNexusEndpointStore()
if err != nil {
return nil, err
}
if f.systemRateLimiter != nil && f.namespaceRateLimiter != nil {
result = persistence.NewNexusEndpointPersistenceRateLimitedClient(result, f.systemRateLimiter, f.namespaceRateLimiter, f.shardRateLimiter, f.logger)
}
result = persistence.NewNexusEndpointPersistenceMetricsClient(result, f.metricsHandler, f.healthSignals, f.logger, f.enableDataLossMetrics)
}
result = persistence.NewNexusEndpointPersistenceRetryableClient(result, retryPolicy, IsPersistenceTransientError)
return result, nil
}
// Close closes this factory
f.dataStoreFactory.Close()
if f.healthSignals != nil {
f.healthSignals.Stop()
}
}
switch err.(type) {
// we retry on DataLoss errors because persistence layer is sometimes unreliable when we immediately read-after-write
case *serviceerror.Unavailable, *serviceerror.DataLoss:
}
if f.metricsHandler == nil && f.healthSignals == nil {
return
}
f.metricsHandler = metrics.NoopMetricsHandler
}
f.healthSignals = persistence.NoopHealthSignalAggregator
}
}
// NewExponentialRetryPolicy returns an instance of ExponentialRetryPolicy using the provided initialInterval
func NewExponentialRetryPolicy(initialInterval time.Duration) *ExponentialRetryPolicy {
retrypolicy.go
p := &ExponentialRetryPolicy{
initialInterval: initialInterval,
backoffCoefficient: defaultBackoffCoefficient,
maximumInterval: defaultMaximumInterval,
expirationInterval: defaultExpirationInterval,
maximumAttempts: defaultMaximumAttempts,
}
return p
}
// NewRetrier is used for creating a new instance of Retrier
return &retrierImpl{
policy: policy,
timeSource: timeSource,
startTime: timeSource.Now(),
currentAttempt: 1,
}
}
// WithInitialInterval sets the initial interval used by ExponentialRetryPolicy for the very first retry
// 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
func (p *ExponentialRetryPolicy) ComputeNextDelay(elapsedTime time.Duration, numAttempts int, _ error) time.Duration {
retrypolicy.go
// Check to see if we ran out of maximum number of attempts
// NOTE: if maxAttempts is X, return done when numAttempts == X, otherwise there will be attempt X+1
if p.maximumAttempts != noMaximumAttempts && numAttempts >= p.maximumAttempts {
return done
}
// Stop retrying after expiration interval is elapsed
return done
}
nextInterval := float64(p.initialInterval) * math.Pow(p.backoffCoefficient, float64(numAttempts-1))
retrypolicy.go
// Disallow retries if initialInterval is negative or nextInterval overflows
if nextInterval <= 0 {
return done
}
}
remainingTime := float64(math.Max(0, float64(p.expirationInterval-elapsedTime)))
retrypolicy.go
nextInterval = math.Min(remainingTime, nextInterval)
}
// Bail out if the next interval is smaller than initial retry interval
if nextDuration < p.initialInterval {
return done
}
return time.Duration(nextInterval)
}
// add jitter to avoid global synchronization
jitterPortion := max(
// Prevent overflow
int(0.2*nextInterval), 1)
nextInterval = nextInterval*0.8 + float64(getJitterRand().Intn(jitterPortion))
return nextInterval
}
func (r *disabledRetryPolicyImpl) ComputeNextDelay(_ time.Duration, _ int, _ error) time.Duration {
// NextBackOff returns the next delay interval. This is used by Retry to delay calling the operation again
nextInterval := r.policy.ComputeNextDelay(r.getElapsedTime(), r.currentAttempt, err)
// Now increment the current attempt
r.currentAttempt++
return nextInterval
}
return r.timeSource.Now().Sub(r.startTime)
}
var _ RetryPolicy = (*ErrorDependentRetryPolicy)(nil)
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 {
}
if r := jitterRand.Load(); r != nil {
}
if !jitterRand.CompareAndSwap(nil, r) {
// Two different goroutines called some top-level
// function at the same time. While the results in
}
r.lk.Lock()
defer r.lk.Unlock()
return r.s.Int63()
}
func (r *RetryLockedSource) Seed(seed int64) {
}
return &RetryLockedSource{
lk: sync.Mutex{},
s: rand.NewSource(time.Now().UnixNano()),
}
}
}
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.
}
*x = ShardInfo{}
mi := &file_temporal_server_api_persistence_v1_executions_proto_msgTypes[0]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *ShardInfo) String() string {
func (*ShardInfo) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_executions_proto_msgTypes[0]
if x != nil {
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
if x != nil {
return x.ShardId
}
return 0
}
if x != nil {
return x.RangeId
}
return 0
}
}
if x != nil {
return x.ReplicationDlqAckLevel
}
return nil
}
if x != nil {
return x.QueueStates
}
return nil
}
}
func init() { file_temporal_server_api_persistence_v1_executions_proto_init() }
executions.pb.go
func file_temporal_server_api_persistence_v1_executions_proto_init() {
if File_temporal_server_api_persistence_v1_executions_proto != nil {
return
}
file_temporal_server_api_persistence_v1_chasm_proto_init()
file_temporal_server_api_persistence_v1_hsm_proto_init()
file_temporal_server_api_persistence_v1_queues_proto_init()
file_temporal_server_api_persistence_v1_update_proto_init()
file_temporal_server_api_persistence_v1_executions_proto_msgTypes[1].OneofWrappers = []any{
(*WorkflowExecutionInfo_LastWorkflowTaskFailureCause)(nil),
(*WorkflowExecutionInfo_LastWorkflowTaskTimedOutType)(nil),
}
file_temporal_server_api_persistence_v1_executions_proto_msgTypes[8].OneofWrappers = []any{
(*TransferTaskInfo_CloseExecutionTaskDetails_)(nil),
(*TransferTaskInfo_ChasmTaskInfo)(nil),
}
file_temporal_server_api_persistence_v1_executions_proto_msgTypes[10].OneofWrappers = []any{
(*VisibilityTaskInfo_ChasmTaskInfo)(nil),
}
file_temporal_server_api_persistence_v1_executions_proto_msgTypes[11].OneofWrappers = []any{
(*TimerTaskInfo_ChasmTaskInfo)(nil),
}
file_temporal_server_api_persistence_v1_executions_proto_msgTypes[13].OneofWrappers = []any{
(*OutboundTaskInfo_StateMachineInfo)(nil),
(*OutboundTaskInfo_ChasmTaskInfo)(nil),
(*OutboundTaskInfo_WorkerCommandsTask)(nil),
}
file_temporal_server_api_persistence_v1_executions_proto_msgTypes[17].OneofWrappers = []any{
(*ActivityInfo_UseWorkflowBuildIdInfo_)(nil),
(*ActivityInfo_LastIndependentlyAssignedBuildId)(nil),
}
file_temporal_server_api_persistence_v1_executions_proto_msgTypes[23].OneofWrappers = []any{
(*Callback_Nexus_)(nil),
(*Callback_Hsm)(nil),
}
file_temporal_server_api_persistence_v1_executions_proto_msgTypes[40].OneofWrappers = []any{
(*ActivityInfo_PauseInfo_Manual_)(nil),
(*ActivityInfo_PauseInfo_RuleId)(nil),
}
file_temporal_server_api_persistence_v1_executions_proto_msgTypes[46].OneofWrappers = []any{
(*CallbackInfo_Trigger_WorkflowClosed)(nil),
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_persistence_v1_executions_proto_rawDesc), len(file_temporal_server_api_persistence_v1_executions_proto_rawDesc)),
NumEnums: 0,
NumMessages: 47,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_persistence_v1_executions_proto_goTypes,
DependencyIndexes: file_temporal_server_api_persistence_v1_executions_proto_depIdxs,
MessageInfos: file_temporal_server_api_persistence_v1_executions_proto_msgTypes,
}.Build()
File_temporal_server_api_persistence_v1_executions_proto = out.File
file_temporal_server_api_persistence_v1_executions_proto_goTypes = nil
file_temporal_server_api_persistence_v1_executions_proto_depIdxs = nil
}
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")
}
case sqlDoubleDollarKeyword[0]:
}
if !hasWordAt(contentStr, sqlIfKeyword, j) {
continue
}
if hasWordsBefore(contentStr, j-1, sqlAddKeyword, sqlColumnKeyword) ||
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
if len(st) == 0 || st[len(st)-1] != sqlIfKeyword[0] {
return nil, errors.New("error reading contents: unmatched `END IF` keyword")
st = st[:len(st)-1]
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])) {
after++
}
}
return hasWordAt(s, word, after), after
}
}
return unicode.IsLetter(rune(c)) || unicode.IsDigit(rune(c))
}
// NewTestLogger returns a logger for tests
// Deprecated: Use testlogger.TestLogger instead.
format := os.Getenv(TestLogFormatEnvVar)
if format == "" {
format = "console"
}
Level: os.Getenv(TestLogLevelEnvVar),
Format: format,
Development: true,
})
// Don't include stack traces for warnings during tests. Only include them for logs with level error and above.
logger = logger.WithOptions(zap.AddStacktrace(zap.ErrorLevel))
return NewZapLogger(logger)
}
// NewZapLogger returns a new zap based logger from zap.Logger
return &zapLogger{
zl: zl,
skip: skipForZapLogger,
baseZl: zl,
}
}
// BuildZapLogger builds and returns a new zap.Logger for this logging configuration
return buildZapLogger(cfg, true)
}
_, path, line, ok := runtime.Caller(skip)
if !ok {
return ""
}
}
fields := make([]zap.Field, len(tags)+1)
l.fillFields(tags, fields)
fields[len(fields)-1] = zap.String(tag.LoggingCallAtKey, caller(l.skip))
return fields
}
// fillFields fill fields parameter with fields read from tags. Optimized for performance.
for i, t := range tags {
fields[i] = zt.Field()
} else {
fields[i] = zap.Any(t.Key(), t.Value())
}
}
if msg == "" {
return defaultMsgForEmpty
}
}
}
if l.zl.Core().Enabled(zap.InfoLevel) {
msg = setDefaultMsg(msg)
fields := l.buildFieldsWithCallAt(tags)
l.zl.Info(msg, fields...)
}
}
}
if l.zl.Core().Enabled(zap.ErrorLevel) {
msg = setDefaultMsg(msg)
fields := l.buildFieldsWithCallAt(tags)
l.zl.Error(msg, fields...)
}
}
}
encodeConfig := DefaultZapEncoderConfig
if disableCaller {
encodeConfig.CallerKey = zapcore.OmitKey
encodeConfig.EncodeCaller = nil
}
if len(cfg.OutputFile) > 0 {
outputPath = cfg.OutputFile
}
outputPath = "stdout"
}
if cfg.Format == "console" {
}
Level: zap.NewAtomicLevelAt(ParseZapLevel(cfg.Level)),
Development: cfg.Development,
Sampling: nil,
Encoding: encoding,
EncoderConfig: encodeConfig,
OutputPaths: []string{outputPath},
ErrorOutputPaths: []string{outputPath},
DisableCaller: disableCaller,
}
logger, _ := config.Build()
return logger
}
}
switch strings.ToLower(level) {
case "debug":
return zap.DebugLevel
case "fatal":
return zap.FatalLevel
return zap.InfoLevel
}
}
func (*Predicate) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[0]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
}
func (*UniversalPredicateAttributes) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[1]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
func (*EmptyPredicateAttributes) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[2]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
func (*AndPredicateAttributes) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[3]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
func (*OrPredicateAttributes) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[4]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
func (*NotPredicateAttributes) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[5]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
func (*NamespaceIdPredicateAttributes) ProtoMessage() {}
func (x *NamespaceIdPredicateAttributes) ProtoReflect() protoreflect.Message {
predicates.pb.go
mi := &file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[6]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
func (*TaskTypePredicateAttributes) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[7]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
func (*DestinationPredicateAttributes) ProtoMessage() {}
func (x *DestinationPredicateAttributes) ProtoReflect() protoreflect.Message {
predicates.pb.go
mi := &file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[8]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
func (*OutboundTaskGroupPredicateAttributes) ProtoMessage() {}
func (x *OutboundTaskGroupPredicateAttributes) ProtoReflect() protoreflect.Message {
predicates.pb.go
mi := &file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[9]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
func (*OutboundTaskPredicateAttributes) ProtoMessage() {}
func (x *OutboundTaskPredicateAttributes) ProtoReflect() protoreflect.Message {
predicates.pb.go
mi := &file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[10]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
}
func init() { file_temporal_server_api_persistence_v1_predicates_proto_init() }
predicates.pb.go
func file_temporal_server_api_persistence_v1_predicates_proto_init() {
if File_temporal_server_api_persistence_v1_predicates_proto != nil {
return
}
file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[0].OneofWrappers = []any{
(*Predicate_UniversalPredicateAttributes)(nil),
(*Predicate_EmptyPredicateAttributes)(nil),
(*Predicate_AndPredicateAttributes)(nil),
(*Predicate_OrPredicateAttributes)(nil),
(*Predicate_NotPredicateAttributes)(nil),
(*Predicate_NamespaceIdPredicateAttributes)(nil),
(*Predicate_TaskTypePredicateAttributes)(nil),
(*Predicate_DestinationPredicateAttributes)(nil),
(*Predicate_OutboundTaskGroupPredicateAttributes)(nil),
(*Predicate_OutboundTaskPredicateAttributes)(nil),
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_persistence_v1_predicates_proto_rawDesc), len(file_temporal_server_api_persistence_v1_predicates_proto_rawDesc)),
NumEnums: 0,
NumMessages: 12,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_persistence_v1_predicates_proto_goTypes,
DependencyIndexes: file_temporal_server_api_persistence_v1_predicates_proto_depIdxs,
MessageInfos: file_temporal_server_api_persistence_v1_predicates_proto_msgTypes,
}.Build()
File_temporal_server_api_persistence_v1_predicates_proto = out.File
file_temporal_server_api_persistence_v1_predicates_proto_goTypes = nil
file_temporal_server_api_persistence_v1_predicates_proto_depIdxs = nil
}
ctx context.Context,
row *sqlplugin.HistoryNodeRow,
// NOTE: txn_id is *= -1 within DB
row.TxnID = -row.TxnID
return mdb.conn.NamedExecContext(ctx,
replaceHistoryNodesQuery,
row,
)
}
// DeleteFromHistoryNode delete a row from history_node table
ctx context.Context,
filter sqlplugin.HistoryNodeSelectFilter,
var query string
if filter.MetadataOnly {
query = getHistoryNodeMetadataQuery
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,
}
}
if err := mdb.conn.SelectContext(ctx, &rows, query, args...); err != nil {
return nil, err
}
}
}
ctx context.Context,
filter sqlplugin.HistoryNodeDeleteFilter,
return mdb.conn.ExecContext(ctx,
deleteHistoryNodesQuery,
filter.ShardID,
filter.TreeID,
filter.BranchID,
filter.MinNodeID,
)
}
// For history_tree table:
ctx context.Context,
row *sqlplugin.HistoryTreeRow,
return mdb.conn.NamedExecContext(ctx,
addHistoryTreeQuery,
row,
)
}
// SelectFromHistoryTree reads one or more rows from history_tree table
ctx context.Context,
filter sqlplugin.HistoryTreeSelectFilter,
var rows []sqlplugin.HistoryTreeRow
err := mdb.conn.SelectContext(ctx,
&rows,
getHistoryTreeQuery,
filter.ShardID,
filter.TreeID,
)
return rows, err
}
// PaginateBranchesFromHistoryTree reads up to page.Limit rows from the history_tree table sorted by their primary key,
ctx context.Context,
page sqlplugin.HistoryTreeBranchPage,
var rows []sqlplugin.HistoryTreeRow
err := mdb.conn.SelectContext(ctx,
&rows,
paginateBranchesQuery,
page.ShardID,
page.TreeID,
page.BranchID,
page.Limit,
)
return rows, err
}
// DeleteFromHistoryTree deletes one or more rows from history_tree table
ctx context.Context,
filter sqlplugin.HistoryTreeDeleteFilter,
return mdb.conn.ExecContext(ctx,
deleteHistoryTreeQuery,
filter.ShardID,
filter.TreeID,
filter.BranchID,
)
}
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
// NewTaskStore returns a new task store
conn, err := f.mainDBConn.Get()
if err != nil {
return nil, err
}
return newTaskPersistence(conn, f.cfg.TaskScanPartitions, f.logger, false, f.serializer)
factory.go
}
// NewFairTaskStore returns a new task store
conn, err := f.mainDBConn.Get()
if err != nil {
return nil, err
}
return newTaskPersistence(conn, f.cfg.TaskScanPartitions, f.logger, true, f.serializer)
factory.go
}
// NewShardStore returns a new shard store
conn, err := f.mainDBConn.Get()
if err != nil {
return nil, err
}
}
// NewMetadataStore returns a new metadata store
conn, err := f.mainDBConn.Get()
if err != nil {
return nil, err
}
}
// NewClusterMetadataStore returns a new ClusterMetadata store
conn, err := f.mainDBConn.Get()
if err != nil {
return nil, err
}
}
// NewExecutionStore returns a new ExecutionStore
conn, err := f.mainDBConn.Get()
if err != nil {
return nil, err
}
}
// NewQueue returns a new queue backed by sql
conn, err := f.mainDBConn.Get()
if err != nil {
return nil, err
}
}
// NewNexusEndpointStore returns a new NexusEndpointStore
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)
}
}
}
// Close closes the underlying connection if the reference count becomes zero
c.Lock()
defer c.Unlock()
c.refCnt--
if c.refCnt == 0 {
return c.DB.Close()
}
}
}
*x = HistoryTreeInfo{}
mi := &file_temporal_server_api_persistence_v1_history_tree_proto_msgTypes[0]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *HistoryTreeInfo) String() string {
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
}
faultInjection *config.FaultInjection,
logger log.Logger,
var result TestCluster
result.logger = logger
result.dbName = dbName
result.schemaDir = schemaDir
result.cfg = config.SQL{
User: username,
Password: password,
ConnectAddr: fmt.Sprintf("%v:%v", host, port),
ConnectProtocol: "tcp",
PluginName: pluginName,
DatabaseName: dbName,
TaskScanPartitions: 4,
ConnectAttributes: connectAttributes,
}
result.faultInjection = faultInjection
return &result
}
// DatabaseName from PersistenceTestCluster interface
// SetupTestDatabase from PersistenceTestCluster interface
s.CreateDatabase()
if s.schemaDir == "" {
return
}
schemaDir := s.schemaDir + "/"
// Config returns the persistence config for connecting to this test cluster
cfg := s.cfg
return config.Persistence{
DefaultStore: "test",
VisibilityStore: "test",
DataStores: map[string]config.DataStore{
"test": {SQL: &cfg, FaultInjection: s.faultInjection},
},
TransactionSizeLimit: dynamicconfig.GetIntPropertyFn(primitives.DefaultTransactionSizeLimit),
}
}
// TearDownTestDatabase from PersistenceTestCluster interface
s.DropDatabase()
}
// CreateDatabase from PersistenceTestCluster interface
cfg2 := s.cfg
// NOTE need to connect with empty name to create new database
if cfg2.PluginName != "sqlite" {
cfg2.DatabaseName = ""
}
defer s.closeAdminDB(db)
err := db.CreateDatabase(s.cfg.DatabaseName)
if err != nil {
panic(err)
}
s.logger.Info("created database", tag.String("database", s.cfg.DatabaseName))
test_sql_persistence.go
}
// DropDatabase from PersistenceTestCluster interface
cfg2 := s.cfg
if cfg2.PluginName == "sqlite" && cfg2.DatabaseName != ":memory:" && cfg2.ConnectAttributes["mode"] != "memory" {
if len(cfg2.DatabaseName) > 3 { // 3 should mean not ., .., empty, or /
// Remove main database file
// NOTE need to connect with empty name to drop the database
db := s.newAdminDB(sqlplugin.DbKindUnknown, &cfg2)
defer s.closeAdminDB(db)
if err := db.DropDatabase(s.cfg.DatabaseName); err != nil {
panic(err)
}
s.logger.Info("dropped database", tag.String("database", s.cfg.DatabaseName))
test_sql_persistence.go
}
}
func (s *TestCluster) newAdminDB(kind sqlplugin.DbKind, cfg *config.SQL) sqlplugin.AdminDB {
test_sql_persistence.go
var db sqlplugin.AdminDB
var err error
err = backoff.ThrottleRetry(
func() error {
db, err = NewSQLAdminDB(kind, cfg, resolver.NewNoopResolver(), log.NewTestLogger(), metrics.NoopMetricsHandler)
return err
},
backoff.NewExponentialRetryPolicy(time.Second).WithExpirationInterval(time.Minute),
nil,
)
s.logger.Fatal("NewSQLAdminDB", tag.Error(err))
}
}
if err := db.Close(); err != nil {
s.logger.Fatal("Close schema DB", tag.Error(err))
}
}
sql.RegisterPlugin(PluginName, &plugin{
queryConverter: &queryConverter{},
connPool: newConnPool(),
})
}
func (p *plugin) GetVisibilityQueryConverter() sqlplugin.VisibilityQueryConverter {
logger log.Logger,
_ metrics.Handler,
conn, err := p.connPool.Allocate(cfg, r, logger, p.createDBConnection)
if err != nil {
return nil, err
}
db.OnClose(func() { p.connPool.Close(cfg) }) // remove reference
return db, nil
}
_ resolver.ServiceResolver,
logger log.Logger,
dsn, err := buildDSN(cfg)
if err != nil {
return nil, fmt.Errorf("error building DSN: %w", err)
}
if err != nil {
return nil, err
}
// respect the user's config values when set, otherwise default to 1 for
// backward compatibility and safety.
if cfg.MaxConns > 0 {
if cfg.MaxConns > 1 && !walEnabled {
logger.Warn(
}
db.SetMaxOpenConns(cfg.MaxConns)
db.SetMaxOpenConns(1)
}
db.SetMaxIdleConns(cfg.MaxIdleConns)
}
db.SetConnMaxLifetime(cfg.MaxConnLifetime)
}
// closes. Set ConnMaxIdleTime to 0 (infinite) to prevent idle connections
// from being reaped, which would destroy the database.
}
// Maps struct names in CamelCase to snake without need for db struct tags.
switch {
// creates temporary DB overlay in order to configure database and schemas
if err := p.setupSQLiteDatabase(cfg, db, logger); err != nil {
_ = db.Close()
return nil, err
}
}
func (p *plugin) setupSQLiteDatabase(cfg *config.SQL, conn *sqlx.DB, logger log.Logger) error {
plugin.go
db := newDB(sqlplugin.DbKindUnknown, cfg.DatabaseName, conn, nil, logger)
defer func() { _ = db.Close() }()
if err != nil {
return err
}
// init tables
}
if cfg.ConnectAttributes == nil {
cfg.ConnectAttributes = make(map[string]string)
}
if err != nil {
return "", err
}
"file:%s?%v",
cfg.DatabaseName,
vals.Encode(),
)
return dsn, nil
}
parameters := url.Values{}
// sort ConnectAttributes to get a deterministic order
keys := expmaps.Keys(cfg.ConnectAttributes)
sort.Strings(keys)
for _, k := range keys {
key := strings.TrimSpace(k)
value := strings.TrimSpace(cfg.ConnectAttributes[k])
if parameters.Get(key) != "" {
return nil, fmt.Errorf("duplicate connection attr: %v:%v, %v:%v",
key,
}
parameters.Set(key, value)
continue
}
}
// set time format
return parameters, nil
}
func (*StateMachineMap) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_hsm_proto_msgTypes[1]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
func (*StateMachineRef) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_hsm_proto_msgTypes[3]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
func (*StateMachineTaskInfo) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_hsm_proto_msgTypes[4]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
func (*StateMachineTimerGroup) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_hsm_proto_msgTypes[5]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
func (*VersionedTransition) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_hsm_proto_msgTypes[6]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
func (*StateMachineTombstoneBatch) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_hsm_proto_msgTypes[7]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
}
func file_temporal_server_api_persistence_v1_hsm_proto_init() {
if File_temporal_server_api_persistence_v1_hsm_proto != nil {
return
}
file_temporal_server_api_persistence_v1_hsm_proto_msgTypes[8].OneofWrappers = []any{
(*StateMachineTombstone_ActivityScheduledEventId)(nil),
(*StateMachineTombstone_TimerId)(nil),
(*StateMachineTombstone_ChildExecutionInitiatedEventId)(nil),
(*StateMachineTombstone_RequestCancelInitiatedEventId)(nil),
(*StateMachineTombstone_SignalExternalInitiatedEventId)(nil),
(*StateMachineTombstone_UpdateId)(nil),
(*StateMachineTombstone_StateMachinePath)(nil),
(*StateMachineTombstone_ChasmNodePath)(nil),
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_persistence_v1_hsm_proto_rawDesc), len(file_temporal_server_api_persistence_v1_hsm_proto_rawDesc)),
NumEnums: 0,
NumMessages: 12,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_persistence_v1_hsm_proto_goTypes,
DependencyIndexes: file_temporal_server_api_persistence_v1_hsm_proto_depIdxs,
MessageInfos: file_temporal_server_api_persistence_v1_hsm_proto_msgTypes,
}.Build()
File_temporal_server_api_persistence_v1_hsm_proto = out.File
file_temporal_server_api_persistence_v1_hsm_proto_goTypes = nil
file_temporal_server_api_persistence_v1_hsm_proto_depIdxs = nil
}
)
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 (e *DeserializationError) IsTerminalTaskError() bool { return true }
func (t *serializerImpl) ShardInfoToBlob(info *persistencespb.ShardInfo) (*commonpb.DataBlob, error) {
serializer.go
return encodeBlob(info, t.encodingType)
}
func (t *serializerImpl) ShardInfoFromBlob(data *commonpb.DataBlob) (*persistencespb.ShardInfo, error) {
serializer.go
shardInfo := &persistencespb.ShardInfo{}
err := Decode(data, shardInfo)
if err != nil {
return nil, err
}
}
}
if queueState.ReaderStates == nil {
queueState.ReaderStates = make(map[int64]*persistencespb.QueueReaderState)
}
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) {
serializer.go
result := &persistencespb.HistoryTreeInfo{}
return result, Decode(data, result)
}
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) {
}
func (t *serializerImpl) QueueMetadataToBlob(metadata *persistencespb.QueueMetadata) (*commonpb.DataBlob, error) {
serializer.go
// TODO change ENCODING_TYPE_JSON to ENCODING_TYPE_PROTO3
return encodeBlob(metadata, enumspb.ENCODING_TYPE_JSON)
}
func (t *serializerImpl) QueueMetadataFromBlob(data *commonpb.DataBlob) (*persistencespb.QueueMetadata, error) {
refreshDuration dynamicconfig.DurationPropertyFn,
logger log.Logger,
if len(clusterInfo) == 0 {
panic("Empty cluster information")
panic("Master cluster name is empty")
panic("Current cluster name is empty")
} else if failoverVersionIncrement == 0 || failoverVersionIncrement > math.MaxInt32 {
metadata.go
panic("Version increment <= 0 or > 2147483647")
}
versionToClusterName, err := updateVersionToClusterName(clusterInfo, failoverVersionIncrement)
metadata.go
if err != nil {
// nolint:forbidigo // matches the other startup-config panics in this constructor
panic(err.Error())
}
panic("Current cluster is not specified in cluster info")
}
panic("Master cluster is not specified in cluster info")
}
maps.Copy(copyClusterInfo, clusterInfo)
if refreshDuration == nil {
refreshDuration = dynamicconfig.GetDurationPropertyFn(refreshInterval)
}
status: common.DaemonStatusInitialized,
enableGlobalNamespace: enableGlobalNamespace,
failoverVersionIncrement: failoverVersionIncrement,
masterClusterName: masterClusterName,
currentClusterName: currentClusterName,
clusterInfo: copyClusterInfo,
versionToClusterName: versionToClusterName,
clusterChangeCallback: make(map[any]CallbackFn),
clusterMetadataStore: clusterMetadataStore,
logger: logger,
refreshDuration: refreshDuration,
}
}
dynamicCollection *dynamicconfig.Collection,
logger log.Logger,
return NewMetadata(
config.EnableGlobalNamespace,
config.FailoverVersionIncrement,
config.MasterClusterName,
config.CurrentClusterName,
config.ClusterInformation,
clusterMetadataStore,
dynamicconfig.ClusterMetadataRefreshInterval.Get(dynamicCollection),
logger,
)
}
func (m *metadataImpl) Start() {
info ClusterInformation,
failoverVersionIncrement int64,
if clusterName == "" {
return errors.New("cluster name must not be empty")
}
return fmt.Errorf("cluster %q: InitialFailoverVersion must be > 0, got %d",
clusterName, info.InitialFailoverVersion)
}
return fmt.Errorf("cluster %q: InitialFailoverVersion (%d) must be < FailoverVersionIncrement (%d)",
clusterName, info.InitialFailoverVersion, failoverVersionIncrement)
}
return fmt.Errorf("cluster %q: RPCAddress must not be empty when Enabled=true", clusterName)
}
}
func updateVersionToClusterName(clusterInfo map[string]ClusterInformation, failoverVersionIncrement int64) (map[int64]string, error) {
metadata.go
versionToClusterName := make(map[int64]string)
for clusterName, info := range clusterInfo {
if err := ValidateClusterInformation(clusterName, info, failoverVersionIncrement); err != nil {
return nil, err
}
return nil, fmt.Errorf(
"duplicate InitialFailoverVersion %d for clusters %q and %q",
info.InitialFailoverVersion, existing, clusterName)
}
}
}
queueType persistence.QueueType,
serializer serialization.Serializer,
queue := &sqlQueue{
SqlStore: NewSQLStore(db, logger, serializer),
queueType: queueType,
logger: logger,
}
return queue, nil
}
func (q *sqlQueue) Init(
ctx context.Context,
blob *commonpb.DataBlob,
if err := q.initializeQueueMetadata(ctx, blob); err != nil {
return err
}
}
}
return -q.queueType
}
func (q *sqlQueue) initializeQueueMetadata(
ctx context.Context,
blob *commonpb.DataBlob,
_, err := q.DB.SelectFromQueueMetadata(ctx, sqlplugin.QueueMetadataFilter{
QueueType: q.queueType,
})
switch err {
case nil:
return nil
result, err := q.DB.InsertIntoQueueMetadata(ctx, &sqlplugin.QueueMetadataRow{
QueueType: q.queueType,
Data: blob.Data,
DataEncoding: blob.EncodingType.String(),
})
if err != nil {
return serviceerror.NewUnavailablef("initializeQueueMetadata operation failed. Error %v", err)
}
if err != nil {
return fmt.Errorf("rowsAffected returned error when initializing queue metadata %v: %v", q.queueType, err)
}
return fmt.Errorf("rowsAffected returned %v queue metadata instead of one", rowsAffected)
}
default:
return err
ctx context.Context,
blob *commonpb.DataBlob,
_, err := q.DB.SelectFromQueueMetadata(ctx, sqlplugin.QueueMetadataFilter{
QueueType: q.getDLQTypeFromQueueType(),
})
switch err {
case nil:
return nil
result, err := q.DB.InsertIntoQueueMetadata(ctx, &sqlplugin.QueueMetadataRow{
QueueType: q.getDLQTypeFromQueueType(),
Data: blob.Data,
DataEncoding: blob.EncodingType.String(),
})
if err != nil {
return serviceerror.NewUnavailablef("initializeDLQMetadata operation failed. Error %v", err)
}
if err != nil {
return fmt.Errorf("rowsAffected returned error when initializing DLQ metadata %v: %v", q.queueType, err)
}
return fmt.Errorf("rowsAffected returned %v DLQ metadata instead of one", rowsAffected)
}
default:
return err
}
func init() { file_temporal_server_api_historyservice_v1_request_response_proto_init() }
request_response.pb.go
func file_temporal_server_api_historyservice_v1_request_response_proto_init() {
if File_temporal_server_api_historyservice_v1_request_response_proto != nil {
return
}
file_temporal_server_api_historyservice_v1_request_response_proto_msgTypes[107].OneofWrappers = []any{
(*StreamWorkflowReplicationMessagesRequest_SyncReplicationState)(nil),
}
file_temporal_server_api_historyservice_v1_request_response_proto_msgTypes[108].OneofWrappers = []any{
(*StreamWorkflowReplicationMessagesResponse_Messages)(nil),
}
file_temporal_server_api_historyservice_v1_request_response_proto_msgTypes[134].OneofWrappers = []any{
(*CompleteNexusOperationChasmRequest_Success)(nil),
(*CompleteNexusOperationChasmRequest_Failure)(nil),
}
file_temporal_server_api_historyservice_v1_request_response_proto_msgTypes[136].OneofWrappers = []any{
(*CompleteNexusOperationRequest_Success)(nil),
(*CompleteNexusOperationRequest_Failure)(nil),
}
file_temporal_server_api_historyservice_v1_request_response_proto_msgTypes[162].OneofWrappers = []any{
(*ExecuteMultiOperationRequest_Operation_StartWorkflow)(nil),
(*ExecuteMultiOperationRequest_Operation_UpdateWorkflow)(nil),
}
file_temporal_server_api_historyservice_v1_request_response_proto_msgTypes[163].OneofWrappers = []any{
(*ExecuteMultiOperationResponse_Response_StartWorkflow)(nil),
(*ExecuteMultiOperationResponse_Response_UpdateWorkflow)(nil),
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_historyservice_v1_request_response_proto_rawDesc), len(file_temporal_server_api_historyservice_v1_request_response_proto_rawDesc)),
NumEnums: 0,
NumMessages: 171,
NumExtensions: 1,
NumServices: 0,
},
GoTypes: file_temporal_server_api_historyservice_v1_request_response_proto_goTypes,
DependencyIndexes: file_temporal_server_api_historyservice_v1_request_response_proto_depIdxs,
MessageInfos: file_temporal_server_api_historyservice_v1_request_response_proto_msgTypes,
ExtensionInfos: file_temporal_server_api_historyservice_v1_request_response_proto_extTypes,
}.Build()
File_temporal_server_api_historyservice_v1_request_response_proto = out.File
file_temporal_server_api_historyservice_v1_request_response_proto_goTypes = nil
file_temporal_server_api_historyservice_v1_request_response_proto_depIdxs = nil
}
}
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
}
)
// 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 {
}
return handler.Timer(d.name)
}
// Operation returns tag for Operation
return NewStringTag("operation", operation)
}
// Error returns tag for Error
return ZapTag{
// NOTE: zap already chosen "error" as key
field: zap.Error(err),
}
}
// ServiceErrorType returns tag for ServiceErrorType
// 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
return errorType(util.ErrorType(err))
}
// 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 (*ChasmTaskInfo) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_chasm_proto_msgTypes[8]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
}
func file_temporal_server_api_persistence_v1_chasm_proto_init() {
if File_temporal_server_api_persistence_v1_chasm_proto != nil {
return
}
file_temporal_server_api_persistence_v1_hsm_proto_init()
file_temporal_server_api_persistence_v1_chasm_proto_msgTypes[1].OneofWrappers = []any{
(*ChasmNodeMetadata_ComponentAttributes)(nil),
(*ChasmNodeMetadata_DataAttributes)(nil),
(*ChasmNodeMetadata_CollectionAttributes)(nil),
(*ChasmNodeMetadata_PointerAttributes)(nil),
}
file_temporal_server_api_persistence_v1_chasm_proto_msgTypes[10].OneofWrappers = []any{
(*ChasmNexusCompletion_Success)(nil),
(*ChasmNexusCompletion_Failure)(nil),
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_persistence_v1_chasm_proto_rawDesc), len(file_temporal_server_api_persistence_v1_chasm_proto_rawDesc)),
NumEnums: 0,
NumMessages: 15,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_persistence_v1_chasm_proto_goTypes,
DependencyIndexes: file_temporal_server_api_persistence_v1_chasm_proto_depIdxs,
MessageInfos: file_temporal_server_api_persistence_v1_chasm_proto_msgTypes,
}.Build()
File_temporal_server_api_persistence_v1_chasm_proto = out.File
file_temporal_server_api_persistence_v1_chasm_proto_goTypes = nil
file_temporal_server_api_persistence_v1_chasm_proto_depIdxs = nil
}
}
func 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{}
}
}
tx *sqlx.Tx,
logger log.Logger,
mdb := &db{
dbKind: dbKind,
dbName: dbName,
onClose: make([]func(), 0),
db: xdb,
tx: tx,
logger: logger,
}
mdb.conn = xdb
if tx != nil {
}
return mdb
}
// BeginTx starts a new transaction and returns a reference to the Tx object
xtx, err := mdb.db.BeginTxx(ctx, nil)
if err != nil {
return nil, err
}
}
// Commit commits a previously started transaction
return mdb.tx.Commit()
}
// Rollback triggers rollback of a previously started transaction
}
mdb.mu.Lock()
mdb.onClose = append(mdb.onClose, hook)
mdb.mu.Unlock()
}
// Close closes the connection to the sqlite db
mdb.mu.RLock()
defer mdb.mu.RUnlock()
for _, hook := range mdb.onClose {
hook()
}
// database connection will be automatically closed by the hook handler when all references are removed
}
func (*UpdateInfo) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_update_proto_msgTypes[3]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
}
func file_temporal_server_api_persistence_v1_update_proto_init() {
if File_temporal_server_api_persistence_v1_update_proto != nil {
return
}
file_temporal_server_api_persistence_v1_hsm_proto_init()
file_temporal_server_api_persistence_v1_update_proto_msgTypes[0].OneofWrappers = []any{
(*UpdateAdmissionInfo_HistoryPointer_)(nil),
}
file_temporal_server_api_persistence_v1_update_proto_msgTypes[3].OneofWrappers = []any{
(*UpdateInfo_Acceptance)(nil),
(*UpdateInfo_Completion)(nil),
(*UpdateInfo_Admission)(nil),
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_persistence_v1_update_proto_rawDesc), len(file_temporal_server_api_persistence_v1_update_proto_rawDesc)),
NumEnums: 0,
NumMessages: 5,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_persistence_v1_update_proto_goTypes,
DependencyIndexes: file_temporal_server_api_persistence_v1_update_proto_depIdxs,
MessageInfos: file_temporal_server_api_persistence_v1_update_proto_msgTypes,
}.Build()
File_temporal_server_api_persistence_v1_update_proto = out.File
file_temporal_server_api_persistence_v1_update_proto_goTypes = nil
file_temporal_server_api_persistence_v1_update_proto_depIdxs = nil
}
// - 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
}
}
// Scan implements sql.Scanner interface to allow this type to be
// parsed transparently by database drivers
if src == nil {
return nil
}
if err := guuid.Scan(src); err != nil {
return err
}
return nil
}
// 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
}
hex.Encode(dst, u[:4])
dst[8] = '-'
hex.Encode(dst[9:13], u[4:6])
dst[13] = '-'
hex.Encode(dst[14:18], u[6:8])
dst[18] = '-'
hex.Encode(dst[19:23], u[8:10])
dst[23] = '-'
hex.Encode(dst[24:], u[10:])
}
// 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
// IsContextCanceledErr checks if the error is context.Canceled or serviceerror.Canceled error
var canceledSvcErr *serviceerror.Canceled
return errors.Is(err, context.Canceled) ||
errors.As(err, &canceledSvcErr)
}
// IsServiceClientTransientError checks if the error is a transient error.
// NewCollection creates a new collection. For subscriptions to work, you must call Start/Stop.
// Get will work without Start/Stop.
// Do this at the first convenient place we have a logger:
logSharedStructureWarnings(logger)
return &Collection{
client: client,
logger: logger,
errCount: -1,
subscriptions: make(map[Key]map[int]any),
convertCache: new(sync.Map),
indexCache: new(sync.Map),
}
}
func (c *Collection) Start() {
// 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
}
}
// 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 NewHistoryBranchUtil(serializer serialization.Serializer) *HistoryBranchUtilImpl {
history_branch_util.go
return &HistoryBranchUtilImpl{
serializer: serializer,
}
}
func (u *HistoryBranchUtilImpl) NewHistoryBranch(
_ time.Duration, // executionTimeout
_ time.Duration, // retentionDuration
var id string
if branchID == nil {
id = *branchID
}
TreeId: treeID,
BranchId: id,
Ancestors: ancestors,
}
data, err := u.serializer.HistoryBranchToBlob(bi)
if err != nil {
return nil, err
}
}
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
}
}
// 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
}
blob, err := codec.NewJSONPBEncoder().Encode(m)
if err != nil {
return nil, err
}
Data: blob,
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)
)
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
}
shardStore ShardStore,
serializer serialization.Serializer,
return &shardManagerImpl{
shardStore: shardStore,
serializer: serializer,
}
}
m.shardStore.Close()
}
func (m *shardManagerImpl) GetName() string {
ctx context.Context,
request *GetOrCreateShardRequest,
createShardInfo := func() (int64, *commonpb.DataBlob, error) {
if shardInfo == nil {
shardInfo = &persistencespb.ShardInfo{}
}
shardInfo.UpdateTime = timestamp.TimeNowPtrUtc()
data, err := m.serializer.ShardInfoToBlob(shardInfo)
if err != nil {
return 0, nil, err
}
}
internalResp, err := m.shardStore.GetOrCreateShard(ctx, &InternalGetOrCreateShardRequest{
shard_manager.go
ShardID: request.ShardID,
CreateShardInfo: createShardInfo,
LifecycleContext: request.LifecycleContext,
})
if err != nil {
return nil, err
}
if err != nil {
return nil, err
}
ShardInfo: shardInfo,
}, nil
}
}
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
logger log.Logger,
create func(*config.SQL, resolver.ServiceResolver, log.Logger) (*sqlx.DB, error),
cp.mu.Lock()
defer cp.mu.Unlock()
dsn, err := buildDSN(cfg)
if err != nil {
return nil, err
}
return entry.db, nil
}
if err != nil {
return nil, err
}
return db, nil
}
// Close virtual connection to database. Only closes for real once no references left.
cp.mu.Lock()
defer cp.mu.Unlock()
dsn, err := buildDSN(cfg)
if err != nil {
return
}
if !ok {
// no such database
return
}
// todo: at the moment pool will persist a single connection to the DB for the whole duration of application
// temporal will start and stop DB connections multiple times, which will cause the loss of the cache
func (*VersionHistoryItem) ProtoMessage() {}
mi := &file_temporal_server_api_history_v1_message_proto_msgTypes[1]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
func (*VersionHistories) ProtoMessage() {}
mi := &file_temporal_server_api_history_v1_message_proto_msgTypes[3]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == 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 (*QueueMetadata) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_queue_metadata_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 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 (*QueueState) ProtoMessage() {}
mi := &file_temporal_server_api_persistence_v1_queues_proto_msgTypes[1]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
}
func file_temporal_server_api_persistence_v1_queues_proto_init() {
if File_temporal_server_api_persistence_v1_queues_proto != nil {
return
}
file_temporal_server_api_persistence_v1_predicates_proto_init()
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_persistence_v1_queues_proto_rawDesc), len(file_temporal_server_api_persistence_v1_queues_proto_rawDesc)),
NumEnums: 0,
NumMessages: 12,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_persistence_v1_queues_proto_goTypes,
DependencyIndexes: file_temporal_server_api_persistence_v1_queues_proto_depIdxs,
MessageInfos: file_temporal_server_api_persistence_v1_queues_proto_msgTypes,
}.Build()
File_temporal_server_api_persistence_v1_queues_proto = out.File
file_temporal_server_api_persistence_v1_queues_proto_goTypes = nil
file_temporal_server_api_persistence_v1_queues_proto_depIdxs = nil
}
logger log.Logger,
serializer serialization.Serializer,
return &sqlShardStore{
SqlStore: NewSQLStore(db, logger, serializer),
currentClusterName: currentClusterName,
}, nil
}
func (m *sqlShardStore) GetClusterName() string {
ctx context.Context,
request *persistence.InternalGetOrCreateShardRequest,
row, err := m.DB.SelectFromShards(ctx, sqlplugin.ShardsFilter{
ShardID: request.ShardID,
})
switch err {
case nil:
return &persistence.InternalGetOrCreateShardResponse{
ShardInfo: persistence.NewDataBlob(row.Data, row.DataEncoding),
}, nil
default:
return nil, serviceerror.NewUnavailablef("GetOrCreateShard: failed to get ShardID %v. Error: %v", request.ShardID, err)
}
return nil, serviceerror.NewNotFoundf("GetOrCreateShard: ShardID %v not found. Error: %v", request.ShardID, err)
}
if err != nil {
return nil, serviceerror.NewUnavailablef("GetOrCreateShard: failed to encode shard info for ShardID %v. Error: %v", request.ShardID, err)
}
ShardID: request.ShardID,
RangeID: rangeID,
Data: shardInfo.Data,
DataEncoding: shardInfo.EncodingType.String(),
}
_, err = m.DB.InsertIntoShards(ctx, row)
if err == nil {
return &persistence.InternalGetOrCreateShardResponse{
ShardInfo: shardInfo,
}, nil
} else if m.DB.IsDupEntryError(err) {
// conflict, try again
request.CreateShardInfo = nil // prevent loop
}
return file_temporal_server_api_enums_v1_common_proto_enumTypes[1].Descriptor()
}
func (ChecksumFlavor) Type() protoreflect.EnumType {
}
return file_temporal_server_api_enums_v1_common_proto_enumTypes[2].Descriptor()
}
func (CallbackState) Type() protoreflect.EnumType {
}
func file_temporal_server_api_enums_v1_common_proto_init() {
if File_temporal_server_api_enums_v1_common_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_enums_v1_common_proto_rawDesc), len(file_temporal_server_api_enums_v1_common_proto_rawDesc)),
NumEnums: 3,
NumMessages: 0,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_enums_v1_common_proto_goTypes,
DependencyIndexes: file_temporal_server_api_enums_v1_common_proto_depIdxs,
EnumInfos: file_temporal_server_api_enums_v1_common_proto_enumTypes,
}.Build()
File_temporal_server_api_enums_v1_common_proto = out.File
file_temporal_server_api_enums_v1_common_proto_goTypes = nil
file_temporal_server_api_enums_v1_common_proto_depIdxs = nil
}
}
return file_temporal_server_api_enums_v1_task_proto_enumTypes[1].Descriptor()
}
func (TaskType) Type() protoreflect.EnumType {
}
return file_temporal_server_api_enums_v1_task_proto_enumTypes[2].Descriptor()
}
func (TaskPriority) Type() protoreflect.EnumType {
}
func file_temporal_server_api_enums_v1_task_proto_init() {
if File_temporal_server_api_enums_v1_task_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_enums_v1_task_proto_rawDesc), len(file_temporal_server_api_enums_v1_task_proto_rawDesc)),
NumEnums: 3,
NumMessages: 0,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_enums_v1_task_proto_goTypes,
DependencyIndexes: file_temporal_server_api_enums_v1_task_proto_depIdxs,
EnumInfos: file_temporal_server_api_enums_v1_task_proto_enumTypes,
}.Build()
File_temporal_server_api_enums_v1_task_proto = out.File
file_temporal_server_api_enums_v1_task_proto_goTypes = nil
file_temporal_server_api_enums_v1_task_proto_depIdxs = nil
}
}
return file_temporal_server_api_enums_v1_workflow_proto_enumTypes[0].Descriptor()
}
func (WorkflowExecutionState) Type() protoreflect.EnumType {
}
return file_temporal_server_api_enums_v1_workflow_proto_enumTypes[1].Descriptor()
}
func (WorkflowBackoffType) Type() protoreflect.EnumType {
}
func file_temporal_server_api_enums_v1_workflow_proto_init() {
if File_temporal_server_api_enums_v1_workflow_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_enums_v1_workflow_proto_rawDesc), len(file_temporal_server_api_enums_v1_workflow_proto_rawDesc)),
NumEnums: 3,
NumMessages: 0,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_enums_v1_workflow_proto_goTypes,
DependencyIndexes: file_temporal_server_api_enums_v1_workflow_proto_depIdxs,
EnumInfos: file_temporal_server_api_enums_v1_workflow_proto_enumTypes,
}.Build()
File_temporal_server_api_enums_v1_workflow_proto = out.File
file_temporal_server_api_enums_v1_workflow_proto_goTypes = nil
file_temporal_server_api_enums_v1_workflow_proto_depIdxs = nil
}
}
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
}
}
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 {
}
if m.DB != nil {
err := m.DB.Close()
if err != nil {
m.logger.Error("Error closing SQL database", tag.Error(err))
}
}
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
}
}
// NewRoute returns a new [Route] instance with the given components.
return Route[T]{components: components}
}
// RouteBuilder is a builder for the [Route] interface.
// NewBuilder creates a new [RouteBuilder] instance, which can be used to define a new [Route] via a fluent API.
return &RouteBuilder[T]{}
}
// With adds a series of [Component] instances to the [Route].
r.components = append(r.components, c...)
return r
}
// Constant adds a [Constant] component to the [Route].
return r.With(Constant[T](values...))
}
// StringVariable adds a [StringVariable] component to the [Route].
func (r *RouteBuilder[T]) StringVariable(name string, getter func(*T) *string) *RouteBuilder[T] {
route.go
return r.With(StringVariable[T](name, getter))
}
// Build returns a read-only [Route].
return NewRoute[T](r.components...)
}
// Representation returns the [github.com/gorilla/mux] compatible string representation of the route for usage in a
// Constant returns a [Component] that represents a series of constant HTTP path components in a Route.
// They will be joined via strings when used to construct a path or path representation.
return values
}
type constant[T any] []string
// StringVariable returns a [Component] that represents a string variable in a Route.
return stringVariable[T]{name, getter}
}
type stringVariable[T any] struct {
func (*VectorClock) ProtoMessage() {}
mi := &file_temporal_server_api_clock_v1_message_proto_msgTypes[0]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
}
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_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 (*BaseExecutionInfo) ProtoMessage() {}
mi := &file_temporal_server_api_workflow_v1_message_proto_msgTypes[2]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
}
func file_temporal_server_api_workflow_v1_message_proto_init() {
if File_temporal_server_api_workflow_v1_message_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_workflow_v1_message_proto_rawDesc), len(file_temporal_server_api_workflow_v1_message_proto_rawDesc)),
NumEnums: 0,
NumMessages: 3,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_workflow_v1_message_proto_goTypes,
DependencyIndexes: file_temporal_server_api_workflow_v1_message_proto_depIdxs,
MessageInfos: file_temporal_server_api_workflow_v1_message_proto_msgTypes,
}.Build()
File_temporal_server_api_workflow_v1_message_proto = out.File
file_temporal_server_api_workflow_v1_message_proto_goTypes = nil
file_temporal_server_api_workflow_v1_message_proto_depIdxs = nil
}
}
func init() { file_temporal_server_chasm_lib_callback_proto_v1_message_proto_init() }
message.pb.go
func file_temporal_server_chasm_lib_callback_proto_v1_message_proto_init() {
if File_temporal_server_chasm_lib_callback_proto_v1_message_proto != nil {
return
}
file_temporal_server_chasm_lib_callback_proto_v1_message_proto_msgTypes[1].OneofWrappers = []any{
message.pb.go
(*Callback_Nexus_)(nil),
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_chasm_lib_callback_proto_v1_message_proto_rawDesc), len(file_temporal_server_chasm_lib_callback_proto_v1_message_proto_rawDesc)),
NumEnums: 1,
NumMessages: 5,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_chasm_lib_callback_proto_v1_message_proto_goTypes,
DependencyIndexes: file_temporal_server_chasm_lib_callback_proto_v1_message_proto_depIdxs,
EnumInfos: file_temporal_server_chasm_lib_callback_proto_v1_message_proto_enumTypes,
MessageInfos: file_temporal_server_chasm_lib_callback_proto_v1_message_proto_msgTypes,
}.Build()
File_temporal_server_chasm_lib_callback_proto_v1_message_proto = out.File
file_temporal_server_chasm_lib_callback_proto_v1_message_proto_goTypes = nil
file_temporal_server_chasm_lib_callback_proto_v1_message_proto_depIdxs = nil
}
}
return file_temporal_server_api_enums_v1_nexus_proto_enumTypes[0].Descriptor()
}
func (NexusOperationState) Type() protoreflect.EnumType {
}
func file_temporal_server_api_enums_v1_nexus_proto_init() {
if File_temporal_server_api_enums_v1_nexus_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_enums_v1_nexus_proto_rawDesc), len(file_temporal_server_api_enums_v1_nexus_proto_rawDesc)),
NumEnums: 1,
NumMessages: 0,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_enums_v1_nexus_proto_goTypes,
DependencyIndexes: file_temporal_server_api_enums_v1_nexus_proto_depIdxs,
EnumInfos: file_temporal_server_api_enums_v1_nexus_proto_enumTypes,
}.Build()
File_temporal_server_api_enums_v1_nexus_proto = out.File
file_temporal_server_api_enums_v1_nexus_proto_goTypes = nil
file_temporal_server_api_enums_v1_nexus_proto_depIdxs = nil
}
}
return file_temporal_server_api_enums_v1_predicate_proto_enumTypes[0].Descriptor()
}
func (PredicateType) Type() protoreflect.EnumType {
}
func file_temporal_server_api_enums_v1_predicate_proto_init() {
if File_temporal_server_api_enums_v1_predicate_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_enums_v1_predicate_proto_rawDesc), len(file_temporal_server_api_enums_v1_predicate_proto_rawDesc)),
NumEnums: 1,
NumMessages: 0,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_enums_v1_predicate_proto_goTypes,
DependencyIndexes: file_temporal_server_api_enums_v1_predicate_proto_depIdxs,
EnumInfos: file_temporal_server_api_enums_v1_predicate_proto_enumTypes,
}.Build()
File_temporal_server_api_enums_v1_predicate_proto = out.File
file_temporal_server_api_enums_v1_predicate_proto_goTypes = nil
file_temporal_server_api_enums_v1_predicate_proto_depIdxs = nil
}
}
return file_temporal_server_api_enums_v1_workflow_task_type_proto_enumTypes[0].Descriptor()
}
func (WorkflowTaskType) Type() protoreflect.EnumType {
}
func init() { file_temporal_server_api_enums_v1_workflow_task_type_proto_init() }
workflow_task_type.pb.go
func file_temporal_server_api_enums_v1_workflow_task_type_proto_init() {
if File_temporal_server_api_enums_v1_workflow_task_type_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_enums_v1_workflow_task_type_proto_rawDesc), len(file_temporal_server_api_enums_v1_workflow_task_type_proto_rawDesc)),
NumEnums: 1,
NumMessages: 0,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_enums_v1_workflow_task_type_proto_goTypes,
DependencyIndexes: file_temporal_server_api_enums_v1_workflow_task_type_proto_depIdxs,
EnumInfos: file_temporal_server_api_enums_v1_workflow_task_type_proto_enumTypes,
}.Build()
File_temporal_server_api_enums_v1_workflow_task_type_proto = out.File
file_temporal_server_api_enums_v1_workflow_task_type_proto_goTypes = nil
file_temporal_server_api_enums_v1_workflow_task_type_proto_depIdxs = nil
}
}
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
}
)
if path := hasSharedStructure(reflect.ValueOf(def), "root"); path != "" {
sharedStructureWarnings.Store(key, path)
}
}
// If you see this warning, it means that a default value used in New*TypedSetting has a
// non-nil slice or map in it. That can lead to confusing behavior since the value from
// dynamic config will be merged over the default value (e.g. the slice will be appended
// to, not replaced). If that behavior is desired, you can avoid this warning by using
// New*TypedSettingWithConverter and referring to dynamicconfig.ConvertStructure
// explicitly. Otherwise use nil slices and maps, including at the top level
// (so `[]string(nil)` instead of `[]string{}`).
logSharedStructureWarningsOnce.Do(func() {
sharedStructureWarnings.Range(func(key, path any) bool {
softassert.Fail(logger,
"default value contains shared structure",
}
// nolint:exhaustive // deliberately not exhaustive
switch v.Kind() {
case reflect.Map, reflect.Slice, reflect.Pointer:
if !v.IsNil() {
return path
}
if !v.IsNil() {
return hasSharedStructure(v.Elem(), path)
}
for i := range v.NumField() {
if p := hasSharedStructure(v.Field(i), path+"."+v.Type().Field(i).Name); p != "" {
return p
}
resolver ReplicationResolver,
mutations ...Mutation,
if resolver == nil {
return nil, serviceerror.NewInvalidArgument("replicationResolver must be provided")
}
info: detail.Info,
config: detail.Config,
configVersion: detail.ConfigVersion,
customSearchAttributesMapper: CustomSearchAttributesMapper{
fieldToAlias: detail.Config.CustomSearchAttributeAliases,
aliasToField: util.InverseMap(detail.Config.CustomSearchAttributeAliases),
},
replicationResolver: resolver,
}
for _, m := range mutations {
}
}
}
return string(id)
}
func (id ID) IsEmpty() bool {
}
return string(n)
}
func (n Name) IsEmpty() bool {
}
func init() { file_temporal_server_api_common_v1_api_category_proto_init() }
api_category.pb.go
func file_temporal_server_api_common_v1_api_category_proto_init() {
if File_temporal_server_api_common_v1_api_category_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_common_v1_api_category_proto_rawDesc), len(file_temporal_server_api_common_v1_api_category_proto_rawDesc)),
NumEnums: 1,
NumMessages: 1,
NumExtensions: 1,
NumServices: 0,
},
GoTypes: file_temporal_server_api_common_v1_api_category_proto_goTypes,
DependencyIndexes: file_temporal_server_api_common_v1_api_category_proto_depIdxs,
EnumInfos: file_temporal_server_api_common_v1_api_category_proto_enumTypes,
MessageInfos: file_temporal_server_api_common_v1_api_category_proto_msgTypes,
ExtensionInfos: file_temporal_server_api_common_v1_api_category_proto_extTypes,
}.Build()
File_temporal_server_api_common_v1_api_category_proto = out.File
file_temporal_server_api_common_v1_api_category_proto_goTypes = nil
file_temporal_server_api_common_v1_api_category_proto_depIdxs = nil
}
}
func init() { file_temporal_server_chasm_lib_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
}
// ThrottleRetry is a resource aware version of Retry.
// Resource exhausted error will be retried using a different throttle retry policy, instead of the specified one.
func ThrottleRetry(operation Operation, policy RetryPolicy, isRetryable IsRetryable) error {
retry.go
ctxOp := func(context.Context) error { return operation() }
return ThrottleRetryContext(context.Background(), ctxOp, policy, isRetryable)
}
policy RetryPolicy,
isRetryable IsRetryable,
var err error
var next time.Duration
if isRetryable == nil {
}
timeSrc := clock.NewRealTimeSource()
r := NewRetrier(policy, timeSrc)
t := NewRetrier(throttleRetryPolicy, timeSrc)
for ctx.Err() == nil {
}
return err
}
}
if _, ok := err.(*serviceerror.ResourceExhausted); ok {
// 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
}
// 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_cluster_v1_message_proto_init() {
if File_temporal_server_api_cluster_v1_message_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_cluster_v1_message_proto_rawDesc), len(file_temporal_server_api_cluster_v1_message_proto_rawDesc)),
NumEnums: 0,
NumMessages: 4,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_cluster_v1_message_proto_goTypes,
DependencyIndexes: file_temporal_server_api_cluster_v1_message_proto_depIdxs,
MessageInfos: file_temporal_server_api_cluster_v1_message_proto_msgTypes,
}.Build()
File_temporal_server_api_cluster_v1_message_proto = out.File
file_temporal_server_api_cluster_v1_message_proto_goTypes = nil
file_temporal_server_api_cluster_v1_message_proto_depIdxs = nil
}
}
func file_temporal_server_api_common_v1_dlq_proto_init() {
if File_temporal_server_api_common_v1_dlq_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_common_v1_dlq_proto_rawDesc), len(file_temporal_server_api_common_v1_dlq_proto_rawDesc)),
NumEnums: 0,
NumMessages: 4,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_common_v1_dlq_proto_goTypes,
DependencyIndexes: file_temporal_server_api_common_v1_dlq_proto_depIdxs,
MessageInfos: file_temporal_server_api_common_v1_dlq_proto_msgTypes,
}.Build()
File_temporal_server_api_common_v1_dlq_proto = out.File
file_temporal_server_api_common_v1_dlq_proto_goTypes = nil
file_temporal_server_api_common_v1_dlq_proto_depIdxs = nil
}
}
func init() { file_temporal_server_api_contextpropagation_v1_message_proto_init() }
message.pb.go
func file_temporal_server_api_contextpropagation_v1_message_proto_init() {
if File_temporal_server_api_contextpropagation_v1_message_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_contextpropagation_v1_message_proto_rawDesc), len(file_temporal_server_api_contextpropagation_v1_message_proto_rawDesc)),
NumEnums: 0,
NumMessages: 2,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_contextpropagation_v1_message_proto_goTypes,
DependencyIndexes: file_temporal_server_api_contextpropagation_v1_message_proto_depIdxs,
MessageInfos: file_temporal_server_api_contextpropagation_v1_message_proto_msgTypes,
}.Build()
File_temporal_server_api_contextpropagation_v1_message_proto = out.File
file_temporal_server_api_contextpropagation_v1_message_proto_goTypes = nil
file_temporal_server_api_contextpropagation_v1_message_proto_depIdxs = nil
}
}
func file_temporal_server_api_deployment_v1_message_proto_init() {
if File_temporal_server_api_deployment_v1_message_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_deployment_v1_message_proto_rawDesc), len(file_temporal_server_api_deployment_v1_message_proto_rawDesc)),
NumEnums: 0,
NumMessages: 75,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_deployment_v1_message_proto_goTypes,
DependencyIndexes: file_temporal_server_api_deployment_v1_message_proto_depIdxs,
MessageInfos: file_temporal_server_api_deployment_v1_message_proto_msgTypes,
}.Build()
File_temporal_server_api_deployment_v1_message_proto = out.File
file_temporal_server_api_deployment_v1_message_proto_goTypes = nil
file_temporal_server_api_deployment_v1_message_proto_depIdxs = nil
}
}
func file_temporal_server_api_enums_v1_cluster_proto_init() {
if File_temporal_server_api_enums_v1_cluster_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_enums_v1_cluster_proto_rawDesc), len(file_temporal_server_api_enums_v1_cluster_proto_rawDesc)),
NumEnums: 2,
NumMessages: 0,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_enums_v1_cluster_proto_goTypes,
DependencyIndexes: file_temporal_server_api_enums_v1_cluster_proto_depIdxs,
EnumInfos: file_temporal_server_api_enums_v1_cluster_proto_enumTypes,
}.Build()
File_temporal_server_api_enums_v1_cluster_proto = out.File
file_temporal_server_api_enums_v1_cluster_proto_goTypes = nil
file_temporal_server_api_enums_v1_cluster_proto_depIdxs = nil
}
}
func file_temporal_server_api_enums_v1_dlq_proto_init() {
if File_temporal_server_api_enums_v1_dlq_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_enums_v1_dlq_proto_rawDesc), len(file_temporal_server_api_enums_v1_dlq_proto_rawDesc)),
NumEnums: 2,
NumMessages: 0,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_enums_v1_dlq_proto_goTypes,
DependencyIndexes: file_temporal_server_api_enums_v1_dlq_proto_depIdxs,
EnumInfos: file_temporal_server_api_enums_v1_dlq_proto_enumTypes,
}.Build()
File_temporal_server_api_enums_v1_dlq_proto = out.File
file_temporal_server_api_enums_v1_dlq_proto_goTypes = nil
file_temporal_server_api_enums_v1_dlq_proto_depIdxs = nil
}
}
func init() { file_temporal_server_api_enums_v1_fairness_state_proto_init() }
fairness_state.pb.go
func file_temporal_server_api_enums_v1_fairness_state_proto_init() {
if File_temporal_server_api_enums_v1_fairness_state_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_enums_v1_fairness_state_proto_rawDesc), len(file_temporal_server_api_enums_v1_fairness_state_proto_rawDesc)),
NumEnums: 1,
NumMessages: 0,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_enums_v1_fairness_state_proto_goTypes,
DependencyIndexes: file_temporal_server_api_enums_v1_fairness_state_proto_depIdxs,
EnumInfos: file_temporal_server_api_enums_v1_fairness_state_proto_enumTypes,
}.Build()
File_temporal_server_api_enums_v1_fairness_state_proto = out.File
file_temporal_server_api_enums_v1_fairness_state_proto_goTypes = nil
file_temporal_server_api_enums_v1_fairness_state_proto_depIdxs = nil
}
}
func file_temporal_server_api_enums_v1_replication_proto_init() {
if File_temporal_server_api_enums_v1_replication_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_enums_v1_replication_proto_rawDesc), len(file_temporal_server_api_enums_v1_replication_proto_rawDesc)),
NumEnums: 3,
NumMessages: 0,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_enums_v1_replication_proto_goTypes,
DependencyIndexes: file_temporal_server_api_enums_v1_replication_proto_depIdxs,
EnumInfos: file_temporal_server_api_enums_v1_replication_proto_enumTypes,
}.Build()
File_temporal_server_api_enums_v1_replication_proto = out.File
file_temporal_server_api_enums_v1_replication_proto_goTypes = nil
file_temporal_server_api_enums_v1_replication_proto_depIdxs = nil
}
}
func file_temporal_server_api_errordetails_v1_message_proto_init() {
if File_temporal_server_api_errordetails_v1_message_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_errordetails_v1_message_proto_rawDesc), len(file_temporal_server_api_errordetails_v1_message_proto_rawDesc)),
NumEnums: 0,
NumMessages: 10,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_errordetails_v1_message_proto_goTypes,
DependencyIndexes: file_temporal_server_api_errordetails_v1_message_proto_depIdxs,
MessageInfos: file_temporal_server_api_errordetails_v1_message_proto_msgTypes,
}.Build()
File_temporal_server_api_errordetails_v1_message_proto = out.File
file_temporal_server_api_errordetails_v1_message_proto_goTypes = nil
file_temporal_server_api_errordetails_v1_message_proto_depIdxs = nil
}
}
func file_temporal_server_api_health_v1_message_proto_init() {
if File_temporal_server_api_health_v1_message_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_health_v1_message_proto_rawDesc), len(file_temporal_server_api_health_v1_message_proto_rawDesc)),
NumEnums: 0,
NumMessages: 3,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_health_v1_message_proto_goTypes,
DependencyIndexes: file_temporal_server_api_health_v1_message_proto_depIdxs,
MessageInfos: file_temporal_server_api_health_v1_message_proto_msgTypes,
}.Build()
File_temporal_server_api_health_v1_message_proto = out.File
file_temporal_server_api_health_v1_message_proto_goTypes = nil
file_temporal_server_api_health_v1_message_proto_depIdxs = nil
}
}
func file_temporal_server_api_historyservice_v1_service_proto_init() {
if File_temporal_server_api_historyservice_v1_service_proto != nil {
return
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_historyservice_v1_service_proto_rawDesc), len(file_temporal_server_api_historyservice_v1_service_proto_rawDesc)),
NumEnums: 0,
NumMessages: 0,
NumExtensions: 0,
NumServices: 1,
},
GoTypes: file_temporal_server_api_historyservice_v1_service_proto_goTypes,
DependencyIndexes: file_temporal_server_api_historyservice_v1_service_proto_depIdxs,
}.Build()
File_temporal_server_api_historyservice_v1_service_proto = out.File
file_temporal_server_api_historyservice_v1_service_proto_goTypes = nil
file_temporal_server_api_historyservice_v1_service_proto_depIdxs = nil
}
}
func init() { file_temporal_server_api_matchingservice_v1_service_proto_init() }
service.pb.go
func file_temporal_server_api_matchingservice_v1_service_proto_init() {
if File_temporal_server_api_matchingservice_v1_service_proto != nil {
return
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_matchingservice_v1_service_proto_rawDesc), len(file_temporal_server_api_matchingservice_v1_service_proto_rawDesc)),
NumEnums: 0,
NumMessages: 0,
NumExtensions: 0,
NumServices: 1,
},
GoTypes: file_temporal_server_api_matchingservice_v1_service_proto_goTypes,
DependencyIndexes: file_temporal_server_api_matchingservice_v1_service_proto_depIdxs,
}.Build()
File_temporal_server_api_matchingservice_v1_service_proto = out.File
file_temporal_server_api_matchingservice_v1_service_proto_goTypes = nil
file_temporal_server_api_matchingservice_v1_service_proto_depIdxs = nil
}
}
func file_temporal_server_api_metrics_v1_message_proto_init() {
if File_temporal_server_api_metrics_v1_message_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_metrics_v1_message_proto_rawDesc), len(file_temporal_server_api_metrics_v1_message_proto_rawDesc)),
NumEnums: 0,
NumMessages: 2,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_metrics_v1_message_proto_goTypes,
DependencyIndexes: file_temporal_server_api_metrics_v1_message_proto_depIdxs,
MessageInfos: file_temporal_server_api_metrics_v1_message_proto_msgTypes,
}.Build()
File_temporal_server_api_metrics_v1_message_proto = out.File
file_temporal_server_api_metrics_v1_message_proto_goTypes = nil
file_temporal_server_api_metrics_v1_message_proto_depIdxs = nil
}
}
func file_temporal_server_api_namespace_v1_message_proto_init() {
if File_temporal_server_api_namespace_v1_message_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_namespace_v1_message_proto_rawDesc), len(file_temporal_server_api_namespace_v1_message_proto_rawDesc)),
NumEnums: 0,
NumMessages: 1,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_namespace_v1_message_proto_goTypes,
DependencyIndexes: file_temporal_server_api_namespace_v1_message_proto_depIdxs,
MessageInfos: file_temporal_server_api_namespace_v1_message_proto_msgTypes,
}.Build()
File_temporal_server_api_namespace_v1_message_proto = out.File
file_temporal_server_api_namespace_v1_message_proto_goTypes = nil
file_temporal_server_api_namespace_v1_message_proto_depIdxs = nil
}
}
func init() { file_temporal_server_api_persistence_v1_chasm_visibility_proto_init() }
chasm_visibility.pb.go
func file_temporal_server_api_persistence_v1_chasm_visibility_proto_init() {
if File_temporal_server_api_persistence_v1_chasm_visibility_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_persistence_v1_chasm_visibility_proto_rawDesc), len(file_temporal_server_api_persistence_v1_chasm_visibility_proto_rawDesc)),
NumEnums: 0,
NumMessages: 2,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_persistence_v1_chasm_visibility_proto_goTypes,
DependencyIndexes: file_temporal_server_api_persistence_v1_chasm_visibility_proto_depIdxs,
MessageInfos: file_temporal_server_api_persistence_v1_chasm_visibility_proto_msgTypes,
}.Build()
File_temporal_server_api_persistence_v1_chasm_visibility_proto = out.File
file_temporal_server_api_persistence_v1_chasm_visibility_proto_goTypes = nil
file_temporal_server_api_persistence_v1_chasm_visibility_proto_depIdxs = nil
}
}
func init() { file_temporal_server_api_persistence_v1_cluster_metadata_proto_init() }
cluster_metadata.pb.go
func file_temporal_server_api_persistence_v1_cluster_metadata_proto_init() {
if File_temporal_server_api_persistence_v1_cluster_metadata_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_persistence_v1_cluster_metadata_proto_rawDesc), len(file_temporal_server_api_persistence_v1_cluster_metadata_proto_rawDesc)),
NumEnums: 0,
NumMessages: 5,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_persistence_v1_cluster_metadata_proto_goTypes,
DependencyIndexes: file_temporal_server_api_persistence_v1_cluster_metadata_proto_depIdxs,
MessageInfos: file_temporal_server_api_persistence_v1_cluster_metadata_proto_msgTypes,
}.Build()
File_temporal_server_api_persistence_v1_cluster_metadata_proto = out.File
file_temporal_server_api_persistence_v1_cluster_metadata_proto_goTypes = nil
file_temporal_server_api_persistence_v1_cluster_metadata_proto_depIdxs = nil
}
}
func init() { file_temporal_server_api_persistence_v1_namespaces_proto_init() }
namespaces.pb.go
func file_temporal_server_api_persistence_v1_namespaces_proto_init() {
if File_temporal_server_api_persistence_v1_namespaces_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_persistence_v1_namespaces_proto_rawDesc), len(file_temporal_server_api_persistence_v1_namespaces_proto_rawDesc)),
NumEnums: 0,
NumMessages: 8,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_persistence_v1_namespaces_proto_goTypes,
DependencyIndexes: file_temporal_server_api_persistence_v1_namespaces_proto_depIdxs,
MessageInfos: file_temporal_server_api_persistence_v1_namespaces_proto_msgTypes,
}.Build()
File_temporal_server_api_persistence_v1_namespaces_proto = out.File
file_temporal_server_api_persistence_v1_namespaces_proto_goTypes = nil
file_temporal_server_api_persistence_v1_namespaces_proto_depIdxs = nil
}
}
func file_temporal_server_api_persistence_v1_tasks_proto_init() {
if File_temporal_server_api_persistence_v1_tasks_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_persistence_v1_tasks_proto_rawDesc), len(file_temporal_server_api_persistence_v1_tasks_proto_rawDesc)),
NumEnums: 0,
NumMessages: 8,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_persistence_v1_tasks_proto_goTypes,
DependencyIndexes: file_temporal_server_api_persistence_v1_tasks_proto_depIdxs,
MessageInfos: file_temporal_server_api_persistence_v1_tasks_proto_msgTypes,
}.Build()
File_temporal_server_api_persistence_v1_tasks_proto = out.File
file_temporal_server_api_persistence_v1_tasks_proto_goTypes = nil
file_temporal_server_api_persistence_v1_tasks_proto_depIdxs = nil
}
}
func file_temporal_server_api_token_v1_message_proto_init() {
if File_temporal_server_api_token_v1_message_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_token_v1_message_proto_rawDesc), len(file_temporal_server_api_token_v1_message_proto_rawDesc)),
NumEnums: 0,
NumMessages: 7,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_token_v1_message_proto_goTypes,
DependencyIndexes: file_temporal_server_api_token_v1_message_proto_depIdxs,
MessageInfos: file_temporal_server_api_token_v1_message_proto_msgTypes,
}.Build()
File_temporal_server_api_token_v1_message_proto = out.File
file_temporal_server_api_token_v1_message_proto_goTypes = nil
file_temporal_server_api_token_v1_message_proto_depIdxs = nil
}
}
func init() { file_temporal_server_api_visibilityservice_v1_request_response_proto_init() }
request_response.pb.go
func file_temporal_server_api_visibilityservice_v1_request_response_proto_init() {
if File_temporal_server_api_visibilityservice_v1_request_response_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_visibilityservice_v1_request_response_proto_rawDesc), len(file_temporal_server_api_visibilityservice_v1_request_response_proto_rawDesc)),
NumEnums: 0,
NumMessages: 5,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_visibilityservice_v1_request_response_proto_goTypes,
DependencyIndexes: file_temporal_server_api_visibilityservice_v1_request_response_proto_depIdxs,
MessageInfos: file_temporal_server_api_visibilityservice_v1_request_response_proto_msgTypes,
}.Build()
File_temporal_server_api_visibilityservice_v1_request_response_proto = out.File
file_temporal_server_api_visibilityservice_v1_request_response_proto_goTypes = nil
file_temporal_server_api_visibilityservice_v1_request_response_proto_depIdxs = nil
}
}
func init() { file_temporal_server_chasm_lib_activity_proto_v1_service_proto_init() }
service.pb.go
func file_temporal_server_chasm_lib_activity_proto_v1_service_proto_init() {
if File_temporal_server_chasm_lib_activity_proto_v1_service_proto != nil {
return
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_chasm_lib_activity_proto_v1_service_proto_rawDesc), len(file_temporal_server_chasm_lib_activity_proto_v1_service_proto_rawDesc)),
NumEnums: 0,
NumMessages: 0,
NumExtensions: 0,
NumServices: 1,
},
GoTypes: file_temporal_server_chasm_lib_activity_proto_v1_service_proto_goTypes,
DependencyIndexes: file_temporal_server_chasm_lib_activity_proto_v1_service_proto_depIdxs,
}.Build()
File_temporal_server_chasm_lib_activity_proto_v1_service_proto = out.File
file_temporal_server_chasm_lib_activity_proto_v1_service_proto_goTypes = nil
file_temporal_server_chasm_lib_activity_proto_v1_service_proto_depIdxs = nil
}
}
func init() { file_temporal_server_chasm_lib_callback_proto_v1_tasks_proto_init() }
tasks.pb.go
func file_temporal_server_chasm_lib_callback_proto_v1_tasks_proto_init() {
if File_temporal_server_chasm_lib_callback_proto_v1_tasks_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_chasm_lib_callback_proto_v1_tasks_proto_rawDesc), len(file_temporal_server_chasm_lib_callback_proto_v1_tasks_proto_rawDesc)),
NumEnums: 0,
NumMessages: 2,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_chasm_lib_callback_proto_v1_tasks_proto_goTypes,
DependencyIndexes: file_temporal_server_chasm_lib_callback_proto_v1_tasks_proto_depIdxs,
MessageInfos: file_temporal_server_chasm_lib_callback_proto_v1_tasks_proto_msgTypes,
}.Build()
File_temporal_server_chasm_lib_callback_proto_v1_tasks_proto = out.File
file_temporal_server_chasm_lib_callback_proto_v1_tasks_proto_goTypes = nil
file_temporal_server_chasm_lib_callback_proto_v1_tasks_proto_depIdxs = nil
}
}
func init() { file_temporal_server_chasm_lib_nexusoperation_proto_v1_service_proto_init() }
service.pb.go
func file_temporal_server_chasm_lib_nexusoperation_proto_v1_service_proto_init() {
if File_temporal_server_chasm_lib_nexusoperation_proto_v1_service_proto != nil {
return
}
file_temporal_server_chasm_lib_nexusoperation_proto_v1_request_response_proto_init()
service.pb.go
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_chasm_lib_nexusoperation_proto_v1_service_proto_rawDesc), len(file_temporal_server_chasm_lib_nexusoperation_proto_v1_service_proto_rawDesc)),
NumEnums: 0,
NumMessages: 0,
NumExtensions: 0,
NumServices: 1,
},
GoTypes: file_temporal_server_chasm_lib_nexusoperation_proto_v1_service_proto_goTypes,
DependencyIndexes: file_temporal_server_chasm_lib_nexusoperation_proto_v1_service_proto_depIdxs,
}.Build()
File_temporal_server_chasm_lib_nexusoperation_proto_v1_service_proto = out.File
file_temporal_server_chasm_lib_nexusoperation_proto_v1_service_proto_goTypes = nil
file_temporal_server_chasm_lib_nexusoperation_proto_v1_service_proto_depIdxs = nil
}
}
func init() { file_temporal_server_chasm_lib_nexusoperation_proto_v1_tasks_proto_init() }
tasks.pb.go
func file_temporal_server_chasm_lib_nexusoperation_proto_v1_tasks_proto_init() {
if File_temporal_server_chasm_lib_nexusoperation_proto_v1_tasks_proto != nil {
return
}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_chasm_lib_nexusoperation_proto_v1_tasks_proto_rawDesc), len(file_temporal_server_chasm_lib_nexusoperation_proto_v1_tasks_proto_rawDesc)),
NumEnums: 0,
NumMessages: 7,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_chasm_lib_nexusoperation_proto_v1_tasks_proto_goTypes,
DependencyIndexes: file_temporal_server_chasm_lib_nexusoperation_proto_v1_tasks_proto_depIdxs,
MessageInfos: file_temporal_server_chasm_lib_nexusoperation_proto_v1_tasks_proto_msgTypes,
}.Build()
File_temporal_server_chasm_lib_nexusoperation_proto_v1_tasks_proto = out.File
file_temporal_server_chasm_lib_nexusoperation_proto_v1_tasks_proto_goTypes = nil
file_temporal_server_chasm_lib_nexusoperation_proto_v1_tasks_proto_depIdxs = nil
}
}
func init() { file_temporal_server_chasm_lib_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.
}
}
metricsHandler metrics.Handler,
logger log.Logger,
blob, err := serializer.QueueMetadataToBlob(
&persistencespb.QueueMetadata{
ClusterAckLevels: make(map[string]int64),
})
if err != nil {
return nil, err
}
if err != nil {
return nil, err
}
queue: queue,
clusterName: clusterName,
metricsHandler: metricsHandler,
logger: logger,
serializer: serializer,
}, nil
}
)
q.queue.Close()
}
func (q *namespaceReplicationQueueImpl) Publish(ctx context.Context, task *replicationspb.ReplicationTask) error {
ctx context.Context,
row *sqlplugin.ShardsRow,
return mdb.conn.ExecContext(ctx,
createShardQry,
row.ShardID,
row.RangeID,
row.Data,
row.DataEncoding,
)
}
// UpdateShards updates one or more rows into shards table
ctx context.Context,
filter sqlplugin.ShardsFilter,
var row sqlplugin.ShardsRow
err := mdb.conn.GetContext(ctx,
&row,
getShardQry,
filter.ShardID,
)
if err != nil {
}
return &row, err
}
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
}
m.persistence.Close()
}
func (m *executionManagerImpl) trimHistoryNode(
enableFairness bool,
serializer serialization.Serializer,
store := SqlStore{
DB: db,
logger: logger,
serializer: serializer,
}
userDataStore := userDataStore{SqlStore: store}
taskQueueStore := taskQueueStore{
SqlStore: store,
version: sqlplugin.MatchingTaskVersion1,
taskScanPartitions: uint32(taskScanPartitions),
}
if enableFairness {
return newTaskManagerV2(db, userDataStore, taskQueueStore, logger, serializer)
}
}
}
return t.field
}
func (t ZapTag) Key() string {
}
return ZapTag{
field: zap.String(key, value),
}
}
func NewStringsTag(key string, value []string) ZapTag {
}
return ZapTag{
field: zap.Bool(key, value),
}
}
func NewErrorTag(key string, value error) ZapTag {
// Shorter helpers (aliases for the New* functions above)
return NewStringTag(key, value)
}
func Strings(key string, value []string) ZapTag {
ctx context.Context,
row *sqlplugin.QueueMetadataRow,
return mdb.conn.NamedExecContext(ctx,
templateCreateQueueMetadataQuery,
row,
)
}
func (mdb *db) UpdateQueueMetadata(
ctx context.Context,
filter sqlplugin.QueueMetadataFilter,
var row sqlplugin.QueueMetadataRow
err := mdb.conn.GetContext(ctx,
&row,
templateGetQueueMetadataQuery,
filter.QueueType,
)
if err != nil {
}
return &row, nil
}
logger log.Logger,
forceRefresh dynamicconfig.BoolPropertyFn,
var saCache atomic.Value
saCache.Store(cache{
searchAttributes: map[string]NameTypeMap{},
dbVersion: 0,
expireOn: time.Time{},
})
return &managerImpl{
logger: logger,
timeSource: timeSource,
cache: saCache,
clusterMetadataManager: clusterMetadataManager,
forceRefresh: forceRefresh,
}
}
// GetSearchAttributes returns all search attributes (including system and build-in) for specified index.
}
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) {
// 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++ {
}
}
}
}
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]
// Otherwise, the type name of the first non-wrapper error in the depth-first traversal of err's tree is returned.
// We consider errors wrapped via [fmt.Errorf], [errors.Join] and some pkg/errors functions to be wrapper errors.
// If any error in the tree has an explicit type name, use it, preferring the first one in the DFS traversal.
var typedErr typedError
if errors.As(err, &typedErr) {
return typedErr.ErrorTypeName()
}
// Special case for context.Cancel error. It is of type errorString, which is not very useful.
return "context.Canceled"
}
// Special case for context.DeadlineExceeded error. It is of unexported type deadlineExceededError.
return "context.DeadlineExceeded"
}
// Otherwise, do a DFS traversal of the error tree, ignoring wrapper errors.
for len(q) > 0 {
err = q[len(q)-1]
q = q[:len(q)-1]
errType := fmt.Sprintf("%T", err)
if !wrapperErrorTypes[errType] {
return strings.TrimPrefix(errType, "*")
}
// The error could implement zero or one of the unary or multi-error wrapper interfaces. It's impossible to
// implement both because they have the same method name. As a result, this is still deterministic.
)
// 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`
}
// NewTestClusterMetadataConfig return an cluster metadata config
func NewTestClusterMetadataConfig(enableGlobalNamespace bool, isMasterCluster bool) *Config {
metadata_test_config.go
masterClusterName := TestCurrentClusterName
if !isMasterCluster {
}
return &Config{
EnableGlobalNamespace: true,
}
EnableGlobalNamespace: false,
FailoverVersionIncrement: TestFailoverVersionIncrement,
MasterClusterName: TestCurrentClusterName,
CurrentClusterName: TestCurrentClusterName,
ClusterInformation: TestSingleDCClusterInfo,
}
}
// GetSQLiteMemoryTestClusterOption return test options
return &TestBaseOptions{
SQLDBPluginName: sqlite.PluginName,
DBName: GenerateRandomDBName(),
DBUsername: testSQLiteUser,
DBPassword: testSQLitePassword,
DBHost: environment.GetLocalhostIP(),
DBPort: 0,
SchemaDir: "",
StoreType: config.StoreTypeSQL,
ConnectAttributes: map[string]string{"mode": testSQLiteMode, "cache": testSQLiteCache},
}
}
)
// 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.
// Timer obtains a timer for the given name.
return NoopTimerMetricFunc
}
// Histogram obtains a histogram for the given name.
}
var NoopGaugeMetricFunc = GaugeFunc(func(f float64, t ...Tag) {})
var NoopHistogramMetricFunc = HistogramFunc(func(i int64, t ...Tag) {})
tracerProvider trace.TracerProvider,
serializer serialization.Serializer,
var dataStoreFactory persistence.DataStoreFactory
defaultStoreCfg := cfg.DataStores[cfg.DefaultStore]
switch {
case defaultStoreCfg.Cassandra != nil:
dataStoreFactory = cassandra.NewFactory(*defaultStoreCfg.Cassandra, r, string(clusterName), logger, metricsHandler, serializer)
dataStoreFactory = sql.NewFactory(*defaultStoreCfg.SQL, r, string(clusterName), logger, metricsHandler, serializer)
case defaultStoreCfg.CustomDataStoreConfig != nil:
dataStoreFactory = abstractDataStoreFactory.NewFactory(*defaultStoreCfg.CustomDataStoreConfig, r, string(clusterName), logger, metricsHandler, serializer)
}
dataStoreFactory = faultinjection.NewFaultInjectionDatastoreFactory(defaultStoreCfg.FaultInjection, dataStoreFactory)
}
if otel.IsEnabled(tracer) {
dataStoreFactory = telemetry.NewTelemetryDataStoreFactory(dataStoreFactory, logger, tracer)
}
}
}
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 {
// Exec executes a sql statement
_, err := mdb.db.Exec(stmt, args...)
return err
}
// ListTables returns a list of tables in this database
// CreateDatabase creates a database if it doesn't exist
// SQLite does not need to create database
return nil
}
// DropDatabase drops a database
// // SQLite does not need to drop database
return nil
}
}
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")
}
}
}()
// dual emit the metric with the all tag. If a blank namespace is provided then
// this converts that to an unknown namespace.
if len(value) == 0 {
}
}
}
return Tag{Key: ErrorTypeTagName, Value: strings.TrimPrefix(util.ErrorType(err), errorPrefix)}
}
func OutcomeTag(outcome string) Tag {
}
return Tag{Key: OperationTagName, Value: value}
}
func StringTag(key string, value string) Tag {
currentClusterName string,
logger log.Logger,
return &clusterMetadataManagerImpl{
serializer: serializer,
persistence: persistence,
currentClusterName: currentClusterName,
logger: logger,
}
}
func (m *clusterMetadataManagerImpl) GetName() string {
}
m.persistence.Close()
}
func (m *clusterMetadataManagerImpl) GetClusterMembers(
logger log.Logger,
clusterName string,
return &metadataManagerImpl{
serializer: serializer,
persistence: persistence,
logger: logger,
clusterName: clusterName,
}
}
func (m *metadataManagerImpl) GetName() string {
}
m.persistence.Close()
}
func (m *metadataManagerImpl) WatchNamespaces(context.Context) (<-chan *NamespaceWatchEvent, error) {
}
d := metricDefinition{
name: name,
description: "",
unit: "",
}
for _, opt := range opts {
opt.apply(&d)
}
return d
}
serializer serialization.Serializer,
logger log.Logger,
return &nexusEndpointManagerImpl{
persistence: persistence,
serializer: serializer,
logger: logger,
}
}
func (m *nexusEndpointManagerImpl) GetName() string {
}
m.persistence.Close()
}
func (m *nexusEndpointManagerImpl) GetNexusEndpoint(
var _ sqlplugin.Plugin = (*plugin)(nil)
sql.RegisterPlugin(PluginName, &plugin{
driver: &driver.PQDriver{},
queryConverter: &queryConverter{},
})
sql.RegisterPlugin(PluginNamePGX, &plugin{
driver: &driver.PGXDriver{},
queryConverter: &queryConverter{},
})
}
func (p *plugin) GetVisibilityQueryConverter() sqlplugin.VisibilityQueryConverter {
)
items := make([]string, len(fields))
for i, field := range fields {
items[i] = fmt.Sprintf("%s = excluded.%s", field, field)
}
return fmt.Sprintf(
// The WHERE clause ensures that no update occurs if the version is behind the saved version.
"ON CONFLICT (namespace_id, run_id) DO UPDATE SET %s WHERE executions_visibility.%s < EXCLUDED.%s",
strings.Join(items, ", "), sqlplugin.VersionColumnName, sqlplugin.VersionColumnName,
)
}
)
items := make([]string, len(fields))
for i, field := range fields {
items[i] = fmt.Sprintf("%s = excluded.%s", field, field)
}
return fmt.Sprintf(
// The WHERE clause ensures that no update occurs if the version is behind the saved version.
"ON CONFLICT (namespace_id, run_id) DO UPDATE SET %s WHERE executions_visibility.%s < EXCLUDED.%s",
strings.Join(items, ", "), sqlplugin.VersionColumnName, sqlplugin.VersionColumnName,
)
}
}
t := reflect.TypeFor[VisibilityRow]()
dbFields := make([]string, t.NumField())
for i := 0; i < t.NumField(); i++ {
f := t.Field(i)
dbFields[i] = f.Tag.Get("db")
if dbFields[i] == "" {
dbFields[i] = strcase.ToSnake(f.Name)
}
}
}
}
return &UnsafeSQLString{Val: val}
}
func NewColName(name string) *ColumnName {
}
func NewSAColumn(alias string, fieldName string, valueType enumspb.IndexedValueType) *SAColumn {
util.go
return &SAColumn{
Alias: alias,
FieldName: fieldName,
ValueType: valueType,
}
}
func NamespaceDivisionSAColumn() *SAColumn {
//
// Note: this function may receive breaking changes or be removed in the future.
statements, err := p.LoadAndSplitQueryFromReaders([]io.Reader{bytes.NewBuffer(executionSchema)})
if err != nil {
return fmt.Errorf("error loading execution schema: %w", err)
}
if err = db.Exec(stmt); err != nil {
return fmt.Errorf("error executing statement %q: %w", stmt, err)
}
}
statements, err = p.LoadAndSplitQueryFromReaders([]io.Reader{bytes.NewBuffer(visibilitySchema)})
setup.go
if err != nil {
return fmt.Errorf("error loading visibility schema: %w", err)
}
if err = db.Exec(stmt); err != nil {
return fmt.Errorf("error executing statement %q: %w", stmt, err)
}
}
}
)
return Category{
id: id,
cType: cType,
name: name,
}
}
func (c Category) ID() int {
}
return c.name
}
func (c Category) Type() CategoryType {
)
return e.Msg
}
func (e *AppendHistoryTimeoutError) Error() string {
// 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
}
}
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), ", ")
}
store TaskStore,
serializer serialization.Serializer,
return &taskManagerImpl{
taskStore: store,
serializer: serializer,
}
}
m.taskStore.Close()
}
func (m *taskManagerImpl) GetName() string {
}
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:
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")
}
}
func GetCallerInfo(
ctx context.Context,
values := GetValues(ctx, CallerNameHeaderName, CallerTypeHeaderName, CallOriginHeaderName)
return CallerInfo{
CallerName: values[0],
CallerType: values[1],
CallOrigin: values[2],
}
}
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) {
// 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 {
// encodingTypeStr not valid, an error will be returned on deserialization
encodingType = enumspb.ENCODING_TYPE_UNSPECIFIED
}
Data: data,
EncodingType: encodingType,
}
}
logger log.Logger,
serializer serialization.Serializer,
return &sqlTaskManagerV1{
SqlStore: NewSQLStore(db, logger, serializer),
userDataStore: uds,
taskQueueStore: tqs,
}, nil
}
func (m *sqlTaskManagerV1) CreateTasks(
logger log.Logger,
serializer serialization.Serializer,
return &sqlTaskManagerV2{
SqlStore: NewSQLStore(db, logger, serializer),
userDataStore: uds,
taskQueueStore: tqs,
}, nil
}
func (m *sqlTaskManagerV2) CreateTasks(
)
// 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
// NewJSONPBEncoder creates a new JSONPBEncoder.
return JSONPBEncoder{}
}
// NewJSONPBIndentEncoder creates a new JSONPBEncoder with indent.
// Encode protobuf struct to bytes.
return e.marshaler.Marshal(pb)
}
// Decode bytes to protobuf struct.
// NewNoopClient returns a Client that has no keys (a Collection using it will always return
// default values).
return StaticClient(nil)
}
// NewNoopCollection creates a new noop collection.
return NewCollection(NewNoopClient(), log.NewNoopLogger())
}
// GetValues returns header values for passed header names.
// It always returns slice of the same size as number of passed header names.
headerValues := make([]string, len(headerNames))
for i, headerName := range headerNames {
if values := metadata.ValueFromIncomingContext(ctx, headerName); len(values) > 0 {
headerValues[i] = values[0]
}
}
}
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)
}
)
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
logger log.Logger,
serializer serialization.Serializer,
return &sqlMetadataManagerV2{
SqlStore: NewSQLStore(db, logger, serializer),
activeClusterName: currentClusterName,
}, nil
}
func (m *sqlMetadataManagerV2) CreateNamespace(
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
logger log.Logger,
serializer serialization.Serializer,
return &sqlClusterMetadataManager{
SqlStore: NewSQLStore(db, logger, serializer),
}, nil
}
logger log.Logger,
serializer serialization.Serializer,
return &sqlNexusEndpointStore{
SqlStore: NewSQLStore(db, logger, serializer),
}, nil
}
func (s *sqlNexusEndpointStore) CreateOrUpdateNexusEndpoint(
var _ sqlplugin.Plugin = (*plugin)(nil)
sql.RegisterPlugin(PluginName, &plugin{
queryConverter: &queryConverter{},
})
}
func (p *plugin) GetVisibilityQueryConverter() sqlplugin.VisibilityQueryConverter {
)
func newNoopSignalAggregator() *noopSignalAggregator { return &noopSignalAggregator{} }
noop_health_signal_aggregator.go
func (a *noopSignalAggregator) Record(_ int32, _ time.Duration, _ error) {}
noop_health_signal_aggregator.go
func (a *noopSignalAggregator) AverageLatency() float64 {
}
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()
}
}
}
_, isNoop := t.(otelnoop.Tracer)
return !isNoop
}
// InverseMap creates the inverse map, ie., for a key-value map, it builds the value-key map.
if m == nil {
return nil
}
invm := make(map[V]K, len(m))
for k, v := range m {
// Size returns the size of the object, in bytes, once serialized
return proto.Size(val)
}
// Equal returns whether two Predicate values are equivalent by recursively
// tasks within the CHASM framework.
// The format of the returned FQN is: "libName.name"
return libName + "." + name
}
// The generated ID is used to uniquely identify components and tasks within the CHASM framework. The same FQN will
// always produce the same ID.
return farm.Fingerprint32([]byte(fqn))
}
// hasBusinessIDAlias returns true if the component has a businessID alias configured
)
return Key{handle: unique.Make(strings.ToLower(s))}
}
func (k Key) String() string {
// NewNoopLogger return a noopLogger
return &noopLogger{}
}
func (n *noopLogger) Debug(string, ...tag.Tag) {}
// NewHostInfoFromAddress creates a new HostInfo instance from a socket address.
return hostAddress(address)
}
// hostAddress is a HostInfo implementation that uses a string as the address and identity.
}
return defaultDataConverter.ToPayload(value)
}
func Decode(p *commonpb.Payload, valuePtr any) error {
// RegisterPlugin adds an auth plugin to the plugin registry
// it is only safe to use from a package init function
translators[name] = plugin
}
func LookupTranslator(name string) (TranslatorPlugin, error) {
baseAPI string,
taskCategory tasks.Category,
return baseAPI + taskCategory.Name()
}
)
return &NoopResolver{}
}
func (c *NoopResolver) Resolve(service string) []string {
}
return ProtoAssertions{t}
}
// ProtoEqual compares two proto messages for equality using proto semantics. Options can be passed to customize
// NewUnprocessableTaskError returns a new UnprocessableTaskError from given message.
return &UnprocessableTaskError{Message: message}
}
func (e UnprocessableTaskError) Error() string {
)
func (c GaugeFunc) Record(v float64, tags ...Tag) { c(v, tags...) }
func (c HistogramFunc) Record(v int64, tags ...Tag) { c(v, tags...) }
)
func newNoopMovingWindowAverage() *noopMovingWindowAverage { return &noopMovingWindowAverage{} }
noop_moving_window_average.go
func (a *noopMovingWindowAverage) Record(_ int64) {}