Atlas › Test
Exact test identity: go.temporal.io/server/common/persistence/tests/TestCassandraHistoryV2Persistence/TestConcurrentlyForkAndAppendBranches
go.temporal.io/server/common/persistence/testsTestCassandraHistoryV2Persistence/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 {
session gocql.Session,
serializer serialization.Serializer,
return &HistoryStore{
Session: session,
HistoryBranchUtil: p.NewHistoryBranchUtil(serializer),
}
}
// AppendHistoryNodes upsert a batch of events as a single node to a history branch
ctx context.Context,
request *p.InternalAppendHistoryNodesRequest,
branchInfo := request.BranchInfo
node := request.Node
if !request.IsNewBranch {
branchInfo.TreeId,
branchInfo.BranchId,
node.NodeID,
node.PrevTransactionID,
node.TransactionID,
node.Events.Data,
node.Events.EncodingType.String(),
).WithContext(ctx)
if err := query.Exec(); err != nil {
return convertTimeoutError(gocql.ConvertError("AppendHistoryNodes", err))
}
}
batch := h.Session.NewBatch(gocql.LoggedBatch).WithContext(ctx)
batch.Query(v2templateInsertTree,
branchInfo.TreeId,
branchInfo.BranchId,
treeInfoDataBlob.Data,
treeInfoDataBlob.EncodingType.String(),
)
batch.Query(v2templateUpsertHistoryNode,
branchInfo.TreeId,
branchInfo.BranchId,
node.NodeID,
node.PrevTransactionID,
node.TransactionID,
node.Events.Data,
node.Events.EncodingType.String(),
)
if err := h.Session.ExecuteBatch(batch); err != nil {
return convertTimeoutError(gocql.ConvertError("AppendHistoryNodes", err))
}
}
ctx context.Context,
request *p.InternalReadHistoryBranchRequest,
branch, err := h.ParseHistoryBranchInfo(request.BranchToken)
if err != nil {
return nil, err
}
if err != nil {
return nil, serviceerror.NewInternalf("ReadHistoryBranch - Gocql TreeId UUID cast failed. Error: %v", err)
}
if err != nil {
return nil, serviceerror.NewInternalf("ReadHistoryBranch - Gocql BranchId UUID cast failed. Error: %v", err)
}
if request.MetadataOnly {
queryString = v2templateReadHistoryNodeMetadata
queryString = v2templateReadHistoryNodeReverse
queryString = v2templateReadHistoryNode
}
query := h.Session.Query(queryString, treeID, branchID, request.MinNodeID, request.MaxNodeID).WithContext(ctx)
history_store.go
iter := query.PageSize(request.PageSize).PageState(request.NextPageToken).Iter()
var pagingToken []byte
if len(iter.PageState()) > 0 {
}
message := make(map[string]any)
for iter.MapScan(message) {
nodes = append(nodes, convertHistoryNode(message))
message = make(map[string]any)
}
return nil, gocql.ConvertError("ReadHistoryBranch", err)
}
Nodes: nodes,
NextPageToken: pagingToken,
}, nil
}
ctx context.Context,
request *p.InternalForkHistoryBranchRequest,
forkB := request.ForkBranchInfo
datablob := request.TreeInfo
cqlTreeID, err := primitives.ValidateUUID(forkB.TreeId)
if err != nil {
return serviceerror.NewInternalf("ForkHistoryBranch - Gocql TreeId UUID cast failed. Error: %v", err)
}
if err != nil {
return serviceerror.NewInternalf("ForkHistoryBranch - Gocql NewBranchID UUID cast failed. Error: %v", err)
}
query := h.Session.Query(v2templateInsertTree, cqlTreeID, cqlNewBranchID, datablob.Data, datablob.EncodingType.String()).WithContext(ctx)
history_store.go
err = query.Exec()
if err != nil {
return gocql.ConvertError("ForkHistoryBranch", err)
}
}
ctx context.Context,
request *p.InternalDeleteHistoryBranchRequest,
batch := h.Session.NewBatch(gocql.LoggedBatch).WithContext(ctx)
batch.Query(v2templateDeleteBranch, request.BranchInfo.TreeId, request.BranchInfo.BranchId)
// delete each branch range
for _, br := range request.BranchRanges {
h.deleteBranchRangeNodes(batch, request.BranchInfo.TreeId, br.BranchId, br.BeginNodeId)
}
if err != nil {
return gocql.ConvertError("DeleteHistoryBranch", err)
}
}
branchID string,
beginNodeID int64,
batch.Query(v2templateRangeDeleteHistoryNode,
treeID,
branchID,
beginNodeID)
}
func (h *HistoryStore) GetAllHistoryTreeBranches(
ctx context.Context,
request *p.GetAllHistoryTreeBranchesRequest,
query := h.Session.Query(v2templateScanAllTreeBranches).WithContext(ctx)
iter := query.PageSize(request.PageSize).PageState(request.NextPageToken).Iter()
var pagingToken []byte
if len(iter.PageState()) > 0 {
pagingToken = iter.PageState()
}
treeUUID := ""
branchUUID := ""
var data []byte
var encoding string
for iter.Scan(&treeUUID, &branchUUID, &data, &encoding) {
TreeID: treeUUID,
BranchID: branchUUID,
Data: data,
Encoding: encoding,
}
branches = append(branches, branch)
treeUUID = ""
branchUUID = ""
data = nil
encoding = ""
}
return nil, gocql.ConvertError("GetAllHistoryTreeBranches", err)
}
Branches: branches,
NextPageToken: pagingToken,
}
return response, nil
}
ctx context.Context,
request *p.InternalGetHistoryTreeContainingBranchRequest,
branch, err := h.ParseHistoryBranchInfo(request.BranchToken)
if err != nil {
return nil, err
}
if err != nil {
return nil, serviceerror.NewInternalf("ReadHistoryBranch. Gocql TreeId UUID cast failed. Error: %v", err)
}
pageSize := 100
var pagingToken []byte
treeInfos := make([]*commonpb.DataBlob, 0, pageSize)
var iter gocql.Iter
for {
iter = query.PageSize(pageSize).PageState(pagingToken).Iter()
pagingToken = iter.PageState()
branchUUID := ""
var data []byte
var encoding string
for iter.Scan(&branchUUID, &data, &encoding) {
treeInfos = append(treeInfos, p.NewDataBlob(data, encoding))
branchUUID = ""
data = []byte{}
encoding = ""
}
return nil, gocql.ConvertError("GetHistoryTree", err)
}
break
}
}
TreeInfos: treeInfos,
}, nil
}
return h.HistoryBranchUtil
}
func convertHistoryNode(
message map[string]any,
nodeID := message["node_id"].(int64)
prevTxnID := message["prev_txn_id"].(int64)
txnID := message["txn_id"].(int64)
var data []byte
var dataEncoding string
if _, ok := message["data"]; ok {
data = message["data"].([]byte)
dataEncoding = message["data_encoding"].(string)
}
return p.InternalHistoryNode{
NodeID: nodeID,
PrevTransactionID: prevTxnID,
TransactionID: txnID,
Events: p.NewDataBlob(data, dataEncoding),
}
}
// 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)
}
}
// NewTestCluster returns a new cassandra test cluster
func NewTestCluster(keyspace, username, password, host string, port int, schemaDir string, faultInjection *config.FaultInjection, logger log.Logger) *TestCluster {
test.go
var result TestCluster
result.logger = logger
result.keyspace = keyspace
if port == 0 {
port = environment.GetCassandraPort()
}
if schemaDir == "" {
schemaDir = testSchemaDir
}
if host == "" {
host = environment.GetCassandraAddress()
}
result.schemaDir = schemaDir
result.cfg = config.Cassandra{
User: username,
Password: password,
Hosts: host,
Port: port,
MaxConns: 2,
ConnectTimeout: 30 * time.Second * debug.TimeoutMultiplier,
Keyspace: keyspace,
}
result.faultInjection = faultInjection
return &result
}
// Config returns the persistence config for connecting to this test cluster
cfg := s.cfg
return config.Persistence{
DefaultStore: "test",
DataStores: map[string]config.DataStore{
"test": {Cassandra: &cfg, FaultInjection: s.faultInjection},
},
TransactionSizeLimit: dynamicconfig.GetIntPropertyFn(primitives.DefaultTransactionSizeLimit),
}
}
// DatabaseName from PersistenceTestCluster interface
return s.keyspace
}
// SetupTestDatabase from PersistenceTestCluster interface
s.CreateSession("system")
s.CreateDatabase()
s.CreateSession(s.DatabaseName())
schemaDir := s.schemaDir + "/"
if !strings.HasPrefix(schemaDir, "/") && !strings.HasPrefix(schemaDir, "../") {
temporalPackageDir := testutils.GetRepoRootDirectory()
schemaDir = path.Join(temporalPackageDir, schemaDir)
}
s.loadSchemaVersion()
}
// TearDownTestDatabase from PersistenceTestCluster interface
s.DropDatabase()
s.session.Close()
}
// CreateSession from PersistenceTestCluster interface
func (s *TestCluster) CreateSession(
keyspace string,
if s.session != nil {
s.session.Close()
}
op := func() error {
session, err := commongocql.NewSession(
func() (*gocql.ClusterConfig, error) {
return commongocql.NewCassandraCluster(
config.Cassandra{
Hosts: s.cfg.Hosts,
Port: s.cfg.Port,
User: s.cfg.User,
Password: s.cfg.Password,
Keyspace: keyspace,
Consistency: &config.CassandraStoreConsistency{
Default: &config.CassandraConsistencySettings{
Consistency: "ONE",
},
},
ConnectTimeout: s.cfg.ConnectTimeout,
},
resolver.NewNoopResolver(),
)
},
log.NewNoopLogger(),
metrics.NoopMetricsHandler,
)
s.session = session
}
return err
}
op,
backoff.NewExponentialRetryPolicy(time.Second).WithExpirationInterval(time.Minute),
nil,
)
if err != nil {
s.logger.Fatal("CreateSession", tag.Error(err))
}
}
// CreateDatabase from PersistenceTestCluster interface
err := CreateCassandraKeyspace(s.session, s.DatabaseName(), 1, true, s.logger)
if err != nil {
s.logger.Fatal("CreateCassandraKeyspace", tag.Error(err))
}
}
// DropDatabase from PersistenceTestCluster interface
err := DropCassandraKeyspace(s.session, s.DatabaseName(), s.logger)
if err != nil && !strings.Contains(err.Error(), "AlreadyExists") {
s.logger.Fatal("DropCassandraKeyspace", tag.Error(err))
}
}
// LoadSchema from PersistenceTestCluster interface
statements, err := p.LoadAndSplitQuery([]string{schemaFile})
if err != nil {
s.logger.Fatal("LoadSchema", tag.Error(err))
}
if err = s.session.Query(stmt).Exec(); err != nil {
s.logger.Fatal("LoadSchema", tag.Error(err))
}
}
}
s.createSchemaVersionTables()
s.updateSchemaVersion(cassandraschema.Version, cassandraschema.Version)
s.writeSchemaUpdateLog("0", cassandraschema.Version, "", "initial version")
s.logger.Info("loaded schema version", tag.String("version", cassandraschema.Version))
}
s.execSchemaVersionQuery(createSchemaVersionTableCQL)
s.execSchemaVersionQuery(createSchemaUpdateHistoryTableCQL)
}
func (s *TestCluster) updateSchemaVersion(newVersion string, minCompatibleVersion string) {
test.go
now := time.Now().UTC()
s.execSchemaVersionQuery(
writeSchemaVersionCQL,
s.keyspace, now, newVersion, minCompatibleVersion)
}
func (s *TestCluster) writeSchemaUpdateLog(oldVersion string, newVersion string, manifestMD5 string, description string) {
test.go
now := time.Now().UTC()
s.execSchemaVersionQuery(
writeSchemaUpdateHistoryCQL,
now.Year(), int(now.Month()), now, oldVersion, newVersion, manifestMD5, description)
}
if err := s.session.Query(stmt, args...).Exec(); err != nil {
s.logger.Fatal("loadSchemaVersion", tag.Error(err))
}
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(
// NewTestBaseWithCassandra returns a persistence test base backed by cassandra datastore
logger := log.NewTestLogger()
testCluster := NewTestClusterForCassandra(options, logger)
return NewTestBaseForCluster(testCluster, logger)
}
func NewTestClusterForCassandra(options *TestBaseOptions, logger log.Logger) *cassandra.TestCluster {
persistence_test_base.go
if options.DBName == "" {
options.DBName = GenerateRandomDBName()
}
testCluster := cassandra.NewTestCluster(options.DBName, options.DBUsername, options.DBPassword, options.DBHost, options.DBPort, options.SchemaDir, options.FaultInjection, logger)
return testCluster
}
}
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) {
// function at the same time. While the results in
// that case are unpredictable, if we just use r here,
// and we are using a seed, we will most likely return
// the same value for both calls. That doesn't seem ideal.
// Just use the first one to get in.
return jitterRand.Load()
}
}
}
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
}
// 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.DebugLevel) {
msg = setDefaultMsg(msg)
fields := l.buildFieldsWithCallAt(tags)
}
if l.zl.Core().Enabled(zap.InfoLevel) {
msg = setDefaultMsg(msg)
fields := l.buildFieldsWithCallAt(tags)
l.zl.Info(msg, fields...)
}
}
}
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
}
func LoadAndSplitQuery(
filePaths []string,
var files []io.Reader
for _, filePath := range filePaths {
f, err := os.Open(filePath)
if err != nil {
return nil, fmt.Errorf("error opening file %s: %w", filePath, err)
}
}
}
func LoadAndSplitQueryFromReaders(
readers []io.Reader,
result := make([]string, 0, querySliceDefaultSize)
for _, r := range readers {
content, err := io.ReadAll(r)
if err != nil {
return nil, fmt.Errorf("error reading contents: %w", err)
}
contentStr := string(bytes.ToLower(content))
for i, j := 0, 0; i < n; i = j {
// stack to keep track of open parenthesis/blocks
var st []byte
var stmtBuilder strings.Builder
stmtLoop:
for ; j < n; j++ {
switch contentStr[j] {
if len(st) == 0 {
j++
break stmtLoop
}
st = append(st, sqlLeftParenthesis)
if len(st) == 0 || st[len(st)-1] != sqlLeftParenthesis {
return nil, fmt.Errorf("error reading contents: unmatched right parenthesis")
}
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) {
st = append(st, sqlBeginKeyword[0])
j += len(sqlBeginKeyword) - 1
}
if !hasWordAt(contentStr, sqlEndKeyword, j) {
continue
}
if ok, after := hasWordAfter(contentStr, sqlIfKeyword, j+len(sqlEndKeyword)); ok {
}
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
}
}
}
return true
}
}
return unicode.IsLetter(rune(c)) || unicode.IsDigit(rune(c))
}
}
*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.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
}
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
}
cfg config.Cassandra,
resolver resolver.ServiceResolver,
var resolvedHosts []string
for _, host := range parseHosts(cfg.Hosts) {
}
if err := ConfigureCassandraCluster(cfg, cluster); err != nil {
return nil, err
}
}
//
//nolint:revive // cognitive complexity 61 (> max enabled 25)
func ConfigureCassandraCluster(cfg config.Cassandra, cluster *gocql.ClusterConfig) error {
client.go
cluster.ProtoVersion = 4
if cfg.Port > 0 {
}
cluster.Authenticator = gocql.PasswordAuthenticator{
Username: cfg.User,
}
}
}
cluster.HostFilter = gocql.DataCentreHostFilter(cfg.Datacenter)
}
if cfg.TLS.CertData != "" && cfg.TLS.CertFile != "" {
return errors.New("only one of certData or certFile properties should be specified")
}
}
if cfg.ConnectTimeout > 0 {
}
if cfg.Timeout > 0 {
cluster.Timeout = cfg.Timeout
}
if cfg.WriteTimeout > 0 {
cluster.WriteTimeout = cfg.WriteTimeout
}
cluster.Consistency = cfg.Consistency.GetConsistency()
cluster.SerialConsistency = cfg.Consistency.GetSerialConsistency()
cluster.DisableInitialHostLookup = cfg.DisableInitialHostLookup
cluster.ReconnectionPolicy = &gocql.ExponentialReconnectionPolicy{
MaxRetries: 30,
InitialInterval: time.Second,
MaxInterval: 10 * time.Second,
}
cluster.PoolConfig.HostSelectionPolicy = gocql.TokenAwareHostPolicy(gocql.RoundRobinHostPolicy())
if cfg.AddressTranslator != nil && cfg.AddressTranslator.Translator != "" {
addressTranslator, err := translator.LookupTranslator(cfg.AddressTranslator.Translator)
if err != nil {
}
}
// parseHosts returns parses a list of hosts separated by comma
var hosts []string
for h := range strings.SplitSeq(input, ",") {
if host := strings.TrimSpace(h); len(host) > 0 {
}
}
}
logger log.Logger,
metricsHandler metrics.Handler,
gocqlSession, err := initSession(logger, newClusterConfigFunc, metricsHandler)
if err != nil {
return nil, err
}
status: common.DaemonStatusStarted,
newClusterConfigFunc: newClusterConfigFunc,
logger: logger,
metricsHandler: metricsHandler,
sessionInitTime: time.Now().UTC(),
}
session.Value.Store(gocqlSession)
return session, nil
}
newClusterConfigFunc func() (*gocql.ClusterConfig, error),
metricsHandler metrics.Handler,
defer log.CapturePanic(logger, &retErr)
cluster, err := newClusterConfigFunc()
if err != nil {
return nil, err
}
defer func() {
metrics.CassandraInitSessionLatency.With(metricsHandler).Record(time.Since(start))
}()
return cluster.CreateSession()
}
stmt string,
values ...any,
q := s.Value.Load().(*gocql.Session).Query(stmt, values...)
if q == nil {
return nil
}
session: s,
gocqlQuery: q,
}
}
func (s *session) NewBatch(
batchType BatchType,
b := s.Value.Load().(*gocql.Session).NewBatch(mustConvertBatchType(batchType))
if b == nil {
return nil
}
session: s,
gocqlBatch: b,
}
}
func (s *session) ExecuteBatch(
b *Batch,
defer func() { s.handleError(retError) }()
}
}
if !atomic.CompareAndSwapInt32(
&s.status,
common.DaemonStatusStarted,
common.DaemonStatusStopped,
) {
}
}
func (s *session) handleError(
err error,
switch err {
case gocql.ErrNoConnections,
gocql.ErrSessionClosed:
s.refresh()
// noop
}
)
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) {
metricsHandler metrics.Handler,
serializer serialization.Serializer,
session, err := commongocql.NewSession(
func() (*gocql.ClusterConfig, error) {
return commongocql.NewCassandraCluster(cfg, r)
},
logger,
metricsHandler,
)
logger.Fatal("unable to initialize cassandra session", tag.Error(err))
}
cfg,
clusterName,
logger,
session,
serializer,
)
}
session commongocql.Session,
serializer serialization.Serializer,
return &Factory{
cfg: cfg,
clusterName: clusterName,
logger: logger,
session: session,
serializer: serializer,
}
}
// NewTaskStore returns a new task store
return NewMatchingTaskStore(f.session, f.logger, false), nil
}
// NewTaskStore returns a new task store
return NewMatchingTaskStore(f.session, f.logger, true), nil
}
// NewShardStore returns a new shard store
return NewShardStore(f.clusterName, f.session, f.logger), nil
}
// NewMetadataStore returns a metadata store
return NewMetadataStore(f.clusterName, f.session, f.logger)
}
// NewClusterMetadataStore returns a metadata store
return NewClusterMetadataStore(f.session, f.logger)
}
// NewExecutionStore returns a new ExecutionStore.
return NewExecutionStore(f.session, f.serializer, f.logger), nil
}
// NewQueue returns a new queue backed by cassandra
return NewQueueStore(queueType, f.session, f.logger)
}
// NewQueueV2 returns a new data-access object for queues and messages stored in Cassandra. It will never return an
// NewNexusEndpointStore returns a new NexusEndpointStore
return NewNexusEndpointStore(f.session, f.logger), nil
}
// Close closes the factory
f.Lock()
defer f.Unlock()
f.session.Close()
}
session gocql.Session,
logger log.Logger,
return &QueueStore{
queueType: queueType,
session: session,
logger: logger,
serializer: serialization.NewSerializer(),
}, nil
}
func (q *QueueStore) Init(
ctx context.Context,
blob *commonpb.DataBlob,
if err := q.initializeQueueMetadata(ctx, blob); err != nil {
return err
}
}
queueType persistence.QueueType,
blob *commonpb.DataBlob,
version := 0
// TODO: remove once cluster_ack_level is removed from DB
clusterAckLevels := map[string]int64{}
query := q.session.Query(templateInsertQueueMetadataQuery,
queueType,
clusterAckLevels,
blob.Data,
blob.EncodingType.String(),
version,
).WithContext(ctx)
_, err := query.MapScanCAS(make(map[string]any))
if err != nil {
return fmt.Errorf("failed to insert initial queue metadata record: %v, Type: %v", err, queueType)
}
// it's ok if the query is not applied, which means that the record exists already.
}
ctx context.Context,
queueType persistence.QueueType,
query := q.session.Query(templateGetQueueMetadataQuery, queueType).WithContext(ctx)
message := make(map[string]any)
err := query.MapScan(message)
if err != nil {
return nil, err
}
return convertQueueMetadata(message, q.serializer)
}
if q.session != nil {
q.session.Close()
}
}
return -q.queueType
}
func (q *QueueStore) initializeQueueMetadata(
ctx context.Context,
blob *commonpb.DataBlob,
_, err := q.getQueueMetadata(ctx, q.queueType)
if gocql.IsNotFoundError(err) {
return q.insertInitialQueueMetadataRecord(ctx, q.queueType, blob)
}
return err
}
ctx context.Context,
blob *commonpb.DataBlob,
_, err := q.getQueueMetadata(ctx, q.getDLQTypeFromQueueType())
if gocql.IsNotFoundError(err) {
return q.insertInitialQueueMetadataRecord(ctx, q.getDLQTypeFromQueueType(), blob)
}
return err
}
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)
}
}
}
session gocql.Session,
logger log.Logger,
return &ShardStore{
ClusterName: clusterName,
Session: session,
Logger: logger,
}
}
func (d *ShardStore) GetOrCreateShard(
ctx context.Context,
request *p.InternalGetOrCreateShardRequest,
query := d.Session.Query(templateGetShardQuery,
request.ShardID,
rowTypeShard,
rowTypeShardNamespaceID,
rowTypeShardWorkflowID,
rowTypeShardRunID,
defaultVisibilityTimestamp,
rowTypeShardTaskID,
).WithContext(ctx)
var data []byte
var encoding string
err := query.Scan(&data, &encoding)
if err == nil {
return &p.InternalGetOrCreateShardResponse{
ShardInfo: p.NewDataBlob(data, encoding),
}, nil
return nil, gocql.ConvertError("GetOrCreateShard", err)
}
// shard was not found and we should create it
if err != nil {
return nil, err
}
request.ShardID,
rowTypeShard,
rowTypeShardNamespaceID,
rowTypeShardWorkflowID,
rowTypeShardRunID,
defaultVisibilityTimestamp,
rowTypeShardTaskID,
shardInfo.Data,
shardInfo.EncodingType.String(),
rangeID,
).WithContext(ctx)
previous := make(map[string]any)
applied, err := query.MapScanCAS(previous)
if err != nil {
return nil, gocql.ConvertError("GetOrCreateShard", err)
}
// conflict, try again
request.CreateShardInfo = nil // prevent loop
return d.GetOrCreateShard(ctx, request)
}
ShardInfo: shardInfo,
}, nil
}
}
if d.Session != nil {
d.Session.Close()
}
}
}
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
}
// 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
// Value returns tag for Value
return NewAnyTag("value", v)
}
// ValueType returns tag for ValueType
)
// 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)
}
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{}
}
}
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
}
session *session,
gocqlQuery *gocql.Query,
return &query{
session: session,
gocqlQuery: gocqlQuery,
}
}
defer func() { q.session.handleError(retError) }()
}
func (q *query) Scan(
dest ...any,
defer func() { q.session.handleError(retError) }()
}
func (q *query) MapScan(
m map[string]any,
defer func() { q.session.handleError(retError) }()
}
func (q *query) MapScanCAS(
dest map[string]any,
defer func() { q.session.handleError(retError) }()
}
iter := q.gocqlQuery.Iter()
return newIter(q.session, iter)
}
q.gocqlQuery.PageSize(n)
return newQuery(q.session, q.gocqlQuery)
}
q.gocqlQuery.PageState(state)
return newQuery(q.session, q.gocqlQuery)
}
func (q *query) Consistency(c Consistency) Query {
// 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
}
// GetConsistency returns the gosql.Consistency setting from the configuration for the given store type
return gocql.ParseConsistency(c.getConsistencySettings().Consistency)
}
// GetSerialConsistency returns the gosql.SerialConsistency setting from the configuration for the store
func (c *CassandraStoreConsistency) GetSerialConsistency() gocql.SerialConsistency {
persistence.go
res, err := parseSerialConsistency(c.getConsistencySettings().SerialConsistency)
if err != nil {
panic(fmt.Sprintf("unable to decode cassandra serial consistency: %v", err))
}
}
func (c *CassandraStoreConsistency) getConsistencySettings() *CassandraConsistencySettings {
persistence.go
return ensureStoreConsistencyNotNil(c).Default
}
func ensureStoreConsistencyNotNil(c *CassandraStoreConsistency) *CassandraStoreConsistency {
persistence.go
if c == nil {
}
}
}
}
}
}
func parseSerialConsistency(serialConsistency string) (gocql.SerialConsistency, error) {
persistence.go
var s gocql.SerialConsistency
err := s.UnmarshalText([]byte(strings.ToUpper(serialConsistency)))
return s, err
}
func (c *SQL) validate() error {
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
}
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
}
}
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
}
// - error if input is malformed
// - input if input can be parsed and is valid
if s == "" {
return "", nil
}
if err != nil {
return "", err
}
}
// NewUUID generates a new random UUID
u, err := guuid.NewV7()
if err != nil {
// Should never happen, but this matches the behavior of google/uuid.NewRandom
return 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:])
}
// NewRoute returns a new [Route] instance with the given components.
return Route[T]{components: components}
}
// RouteBuilder is a builder for the [Route] interface.
// NewBuilder creates a new [RouteBuilder] instance, which can be used to define a new [Route] via a fluent API.
return &RouteBuilder[T]{}
}
// With adds a series of [Component] instances to the [Route].
r.components = append(r.components, c...)
return r
}
// Constant adds a [Constant] component to the [Route].
return r.With(Constant[T](values...))
}
// StringVariable adds a [StringVariable] component to the [Route].
func (r *RouteBuilder[T]) StringVariable(name string, getter func(*T) *string) *RouteBuilder[T] {
route.go
return r.With(StringVariable[T](name, getter))
}
// Build returns a read-only [Route].
return NewRoute[T](r.components...)
}
// Representation returns the [github.com/gorilla/mux] compatible string representation of the route for usage in a
// Constant returns a [Component] that represents a series of constant HTTP path components in a Route.
// They will be joined via strings when used to construct a path or path representation.
return values
}
type constant[T any] []string
// StringVariable returns a [Component] that represents a string variable in a Route.
return stringVariable[T]{name, getter}
}
type stringVariable[T any] struct {
)
// lookup localhost and favor the first ipv4 address
// unless there are only ipv6 addresses available
ips, err := net.LookupIP(domain)
if err != nil || len(ips) == 0 {
// fallback to default instead of error
return localhostIPDefault
}
if ip4 := ip.To4(); ip4 != nil {
}
}
return ips[len(ips)-1].String()
// GetLocalhostIP returns the ip address of the localhost domain
localhostIP := os.Getenv(localhostIPEnv)
ip := net.ParseIP(localhostIP)
if ip != nil {
// if localhost is an ip return it
return ip.String()
}
// otherwise, ignore the value and lookup `localhost`
}
// GetCassandraAddress return the cassandra address
addr := os.Getenv(cassandraSeedsEnv)
if addr == "" {
addr = GetLocalhostIP()
}
return addr
}
// GetCassandraPort return the cassandra port
port := os.Getenv(cassandraPortEnv)
if port == "" {
return cassandraDefaultPort
}
p, err := strconv.Atoi(port)
if err != nil {
func (*VectorClock) ProtoMessage() {}
mi := &file_temporal_server_api_clock_v1_message_proto_msgTypes[0]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
}
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
}
}
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 {
}
return ZapTag{
field: zap.Any(key, value),
}
}
func NewBinaryTag(key string, value []byte) ZapTag {
// Shorter helpers (aliases for the New* functions above)
return NewStringTag(key, value)
}
func Strings(key string, value []string) ZapTag {
}
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 {
}
return &iter{
session: session,
gocqlIter: gocqlIter,
}
}
return it.gocqlIter.Scan(dest...)
}
return it.gocqlIter.MapScan(m)
}
return it.gocqlIter.PageState()
}
defer func() { it.session.handleError(retError) }()
}
var _ p.ExecutionStore = (*ExecutionStore)(nil)
func NewExecutionStore(session gocql.Session, serializer serialization.Serializer, logger log.Logger) *ExecutionStore {
execution_store.go
return &ExecutionStore{
HistoryStore: NewHistoryStore(session, serializer),
MutableStateStore: NewMutableStateStore(session, serializer, logger),
MutableStateTaskStore: NewMutableStateTaskStore(session, serializer),
}
}
func (d *ExecutionStore) CreateWorkflowExecution(
}
if d.HistoryStore.Session != nil {
d.HistoryStore.Session.Close()
}
if d.MutableStateStore.Session != nil {
d.MutableStateStore.Session.Close()
}
if d.MutableStateTaskStore.Session != nil {
d.MutableStateTaskStore.Session.Close()
}
}
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(
// CreateCassandraKeyspace creates the keyspace using this session for given replica count
func CreateCassandraKeyspace(s gocql.Session, keyspace string, replicas int, overwrite bool, logger log.Logger) (err error) {
helpers.go
// if overwrite flag is set, drop the keyspace and create a new one
if overwrite {
err = DropCassandraKeyspace(s, keyspace, logger)
if err != nil {
logger.Error("drop keyspace error", tag.Error(err))
return
}
}
'class' : 'SimpleStrategy', 'replication_factor' : %d}`, keyspace, replicas)).Exec()
if err != nil {
logger.Error("create keyspace error", tag.Error(err))
return
}
return
}
// DropCassandraKeyspace drops the given keyspace, if it exists
func DropCassandraKeyspace(s gocql.Session, keyspace string, logger log.Logger) (err error) {
helpers.go
err = s.Query(fmt.Sprintf("DROP KEYSPACE IF EXISTS %s", keyspace)).Exec()
if err != nil {
logger.Error("drop keyspace error", tag.Error(err))
return
}
return
}
session *session,
gocqlBatch *gocql.Batch,
return &Batch{
session: session,
gocqlBatch: gocqlBatch,
}
}
b.gocqlBatch.Query(stmt, args...)
}
return newBatch(b.session, b.gocqlBatch.WithContext(ctx))
}
func (b *Batch) WithTimestamp(timestamp int64) *Batch {
}
switch batchType {
case LoggedBatch:
return gocql.LoggedBatch
case UnloggedBatch:
return gocql.UnloggedBatch
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.
// 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,
}
}
)
// 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 {
dataStoreFactory = cassandra.NewFactory(*defaultStoreCfg.Cassandra, r, string(clusterName), logger, metricsHandler, serializer)
case defaultStoreCfg.SQL != nil:
dataStoreFactory = sql.NewFactory(*defaultStoreCfg.SQL, 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 {
}
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 {
func newMatchingTaskStoreV1(
session gocql.Session,
return &matchingTaskStoreV1{
Session: session,
userDataStore: userDataStore{Session: session},
taskQueueStore: taskQueueStore{Session: session, version: matchingTaskVersion1},
}
}
// CreateTasks add tasks
}
if d.Session != nil {
d.Session.Close()
}
}
session gocql.Session,
logger log.Logger,
return &MetadataStore{
currentClusterName: currentClusterName,
session: session,
logger: logger,
}, nil
}
// CreateNamespace create a namespace
}
if m.session != nil {
m.session.Close()
}
}
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
}
session gocql.Session,
logger log.Logger,
return &ClusterMetadataStore{
session: session,
logger: logger,
}, nil
}
func (m *ClusterMetadataStore) ListClusterMetadata(
}
if m.session != nil {
m.session.Close()
}
}
session gocql.Session,
logger log.Logger,
return &NexusEndpointStore{
session: session,
logger: logger,
}
}
func (s *NexusEndpointStore) GetName() string {
}
if s.session != nil {
s.session.Close()
}
}
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 {
)
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
}
}
)
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) {
func newMatchingTaskStoreV2(
session gocql.Session,
return &matchingTaskStoreV2{
Session: session,
userDataStore: userDataStore{Session: session},
taskQueueStore: taskQueueStore{Session: session, version: matchingTaskVersion2},
}
}
// CreateTasks add tasks
)
func NewMutableStateStore(session gocql.Session, serializer serialization.Serializer, logger log.Logger) *MutableStateStore {
mutable_state_store.go
return &MutableStateStore{
Session: session,
serializer: serializer,
logger: logger,
}
}
func (d *MutableStateStore) CreateWorkflowExecution(
// 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,
}
}
)
// 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 {
// GetRepoRootDirectory returns the root directory of the temporal repo.
p := &osParams{
Getenv: os.Getenv,
}
for _, opt := range opts {
opt(p)
}
if customRootDirectory := p.Getenv("TEMPORAL_ROOT"); customRootDirectory != "" {
source_root.go
return customRootDirectory
}
}
// 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)
}
)
func NewMutableStateTaskStore(session gocql.Session, serializer serialization.Serializer) *MutableStateTaskStore {
mutable_state_task_store.go
return &MutableStateTaskStore{
Session: session,
serializer: serializer,
}
}
func (d *MutableStateTaskStore) AddHistoryTasks(
)
RegisterTranslator(fixedTranslatorName, NewFixedAddressTranslatorPlugin())
}
type FixedAddressTranslatorPlugin struct {
}
return &FixedAddressTranslatorPlugin{}
}
// GetTranslator What gocql driver does is that it will connect to the first node in the list in configuration
}
sql.RegisterPlugin(PluginName, &plugin{
queryConverter: &queryConverter{},
connPool: newConnPool(),
})
}
func (p *plugin) GetVisibilityQueryConverter() sqlplugin.VisibilityQueryConverter {
)
return &NoopResolver{}
}
return []string{service}
}
func ConvertWeightsToDynamicConfigValue(
weights map[tasks.Priority]int,
weightsForDC := make(map[string]any)
for priority, weight := range weights {
weightsForDC[priority.String()] = weight
}
return weightsForDC
}
}
return Key{
FireTime: fireTime,
TaskID: taskID,
}
}
func ValidateKey(key Key) error {
}
return backoff.NewExponentialRetryPolicy(cfg.InitialInterval).
WithMaximumInterval(cfg.MaxInterval).
WithExpirationInterval(backoff.NoInterval)
}
var defaultRetryPolicyConfig = RetryPolicyConfig{
// serializeConflictToken serializes a conflict token as a byte slice.
token := make([]byte, 8)
binary.LittleEndian.PutUint64(token, uint64(conflictToken))
return token
}
// newTaggedLogger returns a logger tagged with the Scheduler's attributes.
// register adds a metric definition to the list of pending metric definitions. This method is thread-safe.
c.Lock()
defer c.Unlock()
c.definitions = append(c.definitions, d)
}
// buildCatalog builds a catalog from the list of pending metric definitions. It is safe to call this method multiple
logger log.Logger,
enableFairness bool,
if enableFairness {
}
}
var _ sqlplugin.Plugin = (*plugin)(nil)
sql.RegisterPlugin(PluginName, &plugin{
queryConverter: &queryConverter{},
})
}
func (p *plugin) GetVisibilityQueryConverter() sqlplugin.VisibilityQueryConverter {
}
return &connPool{
pool: make(map[string]entry),
}
}
// Allocate allocates the shared database in the pool or returns already exists instance with the same DSN. If instance
)
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 {
}
return errors.Is(err, gocql.ErrNotFound)
}
// RegisterPlugin adds an auth plugin to the plugin registry
// it is only safe to use from a package init function
translators[name] = plugin
}
func LookupTranslator(name string) (TranslatorPlugin, error) {
baseAPI string,
taskCategory tasks.Category,
return baseAPI + taskCategory.Name()
}
// RegisterPlugin will register a SQL plugin
if _, ok := supportedPlugins[pluginName]; ok {
panic("plugin " + pluginName + " already registered")
}
}
}
return 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 {
// We have to use pointer is because in golang: "recover return nil if was not called directly by a deferred function."
// And we have to set the returned error otherwise our handler will return nil as error which is incorrect
if panicObj := recover(); panicObj != nil {
err, ok := panicObj.(error)
if !ok {
)
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) {}