Frontier kind: Code frontier
unlabeled · c_e691d36f35a5
2 tests · 6495 LOC · 221 files · introduces 0 tests · 176 LOC · 4 files
The orange circle is the focus. Violet and green circles are every ancestor and descendant, broader and narrower, at any distance; blue squares and pink diamonds are the introduced files and exact introduced tests of every visible concept, not only the focus's. Arrows point from broader to narrower concepts and bridge only concepts omitted from this view. Undirected links show source or test introduction. Concept and file size follows LOC; exact test nodes use test-count units.
Introduced files, introduced tests, and structurally relevant concept specialization
In the embedded map, ordinary wheel input scrolls the page; use the visible controls to zoom and drag to pan. Open the full-screen map for canvas navigation: wheel pans, Ctrl/Command plus wheel zooms, and arrow keys pan when this region is focused. On touch screens, open the full-screen map to pan or pinch. If JavaScript or WebGL is unavailable, use the native relationship evidence on this page.
Graph controls are ready.
Interactive rendering requires JavaScript and WebGL. Use the native relationship evidence on this page while the interactive map is unavailable.
Every exact file and test below is linked only from the concept that introduces it.
go.temporal.io/server/common/persistence/client/TestQuotasSuite/TestAPITypeCallOriginPriorityOverrideMappinggo.temporal.io/server/common/persistence/client/TestQuotasSuite/TestBackgroundTypeAPIPriorityOverrideMappinggo.temporal.io/server/common/persistence/client/TestQuotasSuite/TestCallerTypeDefaultPriorityMappinggo.temporal.io/server/common/persistence/client/TestQuotasSuite/TestRequestPrioritiesOrderedEvery collected test enters the hierarchy at exactly one concept.
No tests are introduced at this concept. Its intent tests are introduced by other concepts.
Every collected source range enters the hierarchy at exactly one concept.
4 files ranked by introduced lines: 176 introduced LOC across 35 ranges. Expand a file to inspect source; the > gutter marks introduced lines.
ms, err := wfCtx.LoadMutableState(ctx, r.shardContext)
switch err.(type) {
// no-op, continue to replicate workflow state
case nil:
return err
}
skipCloseTransferTask := request.GetIsForceReplication() && request.GetIsCloseTransferTaskAcked()
workflow_state_replicator.go
return r.applySnapshotWhenWorkflowNotExist(
ctx,
namespaceID,
wid,
rid,
chasm.WorkflowArchetypeID,
wfCtx,
releaseFn,
request.GetWorkflowState(),
request.RemoteCluster,
nil,
false,
skipCloseTransferTask,
)
}
isStateBased bool,
skipGenerateCloseTransferTask bool,
var lastWriteVersion int64
executionInfo := sourceMutableState.ExecutionInfo
if transitionHistory := executionInfo.GetTransitionHistory(); len(transitionHistory) != 0 {
lastWriteVersion = transitionhistory.LastVersionedTransition(transitionHistory).NamespaceFailoverVersion
// TODO: remove following logic once transition history is fully enabled.
currentVersionHistory, err := versionhistory.GetCurrentVersionHistory(executionInfo.VersionHistories)
if err != nil {
return err
}
lastEventItem, err := versionhistory.GetLastVersionHistoryItem(currentVersionHistory)
workflow_state_replicator.go
if err != nil {
return err
}
}
if err != nil {
return err
}
r.shardContext,
r.shardContext.GetEventsCache(),
r.logger,
ns,
sourceMutableState,
lastWriteVersion,
)
if err != nil {
return err
}
ctx,
sourceCluster,
namespaceID,
workflowID,
runID,
mutableState,
isStateBased,
); err != nil {
return err
}
err = r.createNewRunWorkflow(
ctx,
}
err = taskRefresher.Refresh(ctx, mutableState, skipGenerateCloseTransferTask)
if err != nil {
return err
}
ctx,
archtypeID,
NewWorkflow(
r.clusterMetadata,
wfCtx,
mutableState,
releaseFn,
),
)
}
mutableState *workflow.MutableStateImpl,
isStateBased bool,
versionHistories := mutableState.GetExecutionInfo().VersionHistories
isEmpty, err := versionhistory.IsCurrentVersionHistoryEmpty(versionHistories)
if err != nil || isEmpty {
return err
}
currentVersionHistory, err := versionhistory.GetCurrentVersionHistory(versionHistories)
workflow_state_replicator.go
if err != nil {
return err
}
lastEventItem, err := versionhistory.GetLastVersionHistoryItem(currentVersionHistory)
workflow_state_replicator.go
if err != nil {
return err
}
// The following sanitizes the branch token from the source cluster to this target cluster by re-initializing it.
branchInfo, err := r.shardContext.GetExecutionManager().GetHistoryBranchUtil().ParseHistoryBranchInfo(
workflow_state_replicator.go
currentVersionHistory.GetBranchToken(),
)
if err != nil {
return err
}
if archetypeID != chasm.WorkflowArchetypeID {
return softassert.UnexpectedInternalErr(
r.logger,
}
backfillBranchToken, err := r.shardContext.GetExecutionManager().GetHistoryBranchUtil().NewHistoryBranch(
workflow_state_replicator.go
namespaceID.String(),
workflowID,
runID,
branchInfo.GetTreeId(),
&branchInfo.BranchId,
branchInfo.Ancestors,
time.Duration(0),
time.Duration(0),
time.Duration(0),
)
if err != nil {
return err
}
// Use the history tree id to be the original run id.
// https://github.com/temporalio/temporal/issues/6501
if runID != originalRunID {
// At this point, it already acquired the workflow lock on the run ID.
// Get the lock of root run id to make sure no concurrent backfill history across multiple runs.
// Get the last batch node id to check if the history data is already in DB.
localHistoryIterator := collection.NewPagingIterator(r.getHistoryFromLocalPaginationFn(
workflow_state_replicator.go
ctx,
backfillBranchToken,
lastEventItem.EventId,
))
var lastBatchNodeID int64
for localHistoryIterator.HasNext() {
localHistoryBatch, err := localHistoryIterator.Next()
switch err.(type) {
case nil:
if len(localHistoryBatch.GetEvents()) > 0 {
}
remoteHistoryIterator := collection.NewPagingIterator(r.getHistoryFromRemotePaginationFn(
workflow_state_replicator.go
ctx,
remoteClusterName,
namespaceID,
workflowID,
runID,
common.EmptyEventID,
common.EmptyVersion,
lastEventItem.EventId+1,
lastEventItem.Version),
)
historyBranchUtil := r.executionMgr.GetHistoryBranchUtil()
historyBranch, err := historyBranchUtil.ParseHistoryBranchInfo(backfillBranchToken)
if err != nil {
return err
}
ns, err := r.namespaceRegistry.GetNamespaceByID(namespaceID)
if err == nil && ns != nil {
nsName = ns.Name().String()
}
// In standby cluster, use a background low priority which is higher than the standby task processing
if ns.ActiveClusterName(namespace.RoutingKey{ID: workflowID}) == r.clusterMetadata.GetCurrentClusterName() {
// In active cluster, use lowest priority to minimize the impact to live traffic
callerType = headers.CallerTypePreemptable
}
"AppendRawHistoryNodes",
1,
nsName,
callerType,
0,
"",
)
prevTxnID := common.EmptyEventTaskID
var prevBranchID string
sortedAncestors := sortAncestors(historyBranch.GetAncestors())
sortedAncestorsIdx := 0
var ancestors []*persistencespb.HistoryBranchRange
BackfillLoop:
for remoteHistoryIterator.HasNext() {
historyBlob, err := remoteHistoryIterator.Next()
if err != nil {
}
return mutableState.SetCurrentBranchToken(backfillBranchToken)
}
branchToken []byte,
lastEventID int64,
return func(paginationToken []byte) ([]*historypb.History, []byte, error) {
response, err := r.executionMgr.ReadHistoryBranchByBatch(ctx, &persistence.ReadHistoryBranchRequest{
ShardID: r.shardContext.GetShardID(),
BranchToken: branchToken,
MinEventID: common.FirstEventID,
MaxEventID: lastEventID + 1,
PageSize: 100,
NextPageToken: paginationToken,
})
if err != nil {
return nil, nil, err
}
}
func sortAncestors(ans []*persistencespb.HistoryBranchRange) []*persistencespb.HistoryBranchRange {
workflow_state_replicator.go
if len(ans) > 0 {
// sort ans based onf EndNodeID so that we can set BeginNodeID
sort.Slice(ans, func(i, j int) bool { return ans[i].GetEndNodeId() < ans[j].GetEndNodeId() })
mutableStateRecord *persistencespb.WorkflowMutableState,
lastWriteVersion int64,
// Although new versions of temporal server will perform state sanitization,
// we have to keep the sanitization logic here as well for backward compatibility in case
// source cluster is running an old version and doesn't do the sanitization.
SanitizeMutableState(mutableStateRecord)
if err := common.DiscardUnknownProto(mutableStateRecord); err != nil {
return nil, err
}
mutableState, err := NewMutableStateFromDB(shard, eventsCache, logger, namespaceEntry, mutableStateRecord, 1)
mutable_state_impl.go
if err != nil {
return nil, err
}
return mutableState, nil
}
func (ms *MutableStateImpl) GetWorkflowCloseTime(ctx context.Context) (time.Time, error) {
if ms.executionState.GetState() == enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED && ms.executionInfo.CloseTime == nil {
// This is for backward compatible. Prior to v1.16 does not have close time in mutable state (Added by 05/21/2022).
mutable_state_impl.go
// TODO: remove this logic when all mutable state contains close time.
completionEvent, err := ms.GetCompletionEvent(ctx)
if err != nil {
return time.Time{}, err
}
}
return ms.executionInfo.CloseTime.AsTime(), nil
}
func (x *ReplicateWorkflowStateRequest) GetIsForceReplication() bool {
request_response.pb.go
if x != nil {
return x.IsForceReplication
}
return false
}
}
if x != nil {
return x.Ancestors
}
return nil
}