Vihren Atlas

workflow_state_replicator.go ×21

Code frontier

unlabeled · c_77da5c2ffe40

4 tests · 4621 LOC · 200 files · introduces 0 tests · 105 LOC · 3 files

Introduces — evidence that enters the hierarchy at this concept

Code
26 ranges105 lines · 3 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
729 ranges4621 lines · 200 files · Browse complete extent
All tests (intent)
4 testsBrowse complete intent

Neighbourhood graph

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.

Native relationship evidence

Every exact file and test below is linked only from the concept that introduces it.

Introduced tests

Every collected test enters the hierarchy at exactly one concept.

No tests are introduced at this concept. Its intent tests are introduced by other concepts.

Introduced code

Every collected source range enters the hierarchy at exactly one concept.

3 files ranked by introduced lines: 105 introduced LOC across 26 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/service/history/ndc/workflow_state_replicator.go 82 introduced LOC · 21 ranges

Open complete file

1298 }
1299
1300 > startEventID := localLastItem.GetEventId() // exclusive workflow_state_replicator.go
1301 > startEventVersion := localLastItem.GetVersion()
1302 > endEventID := sourceLastItem.GetEventId() // inclusive
1303 > endEventVersion := sourceLastItem.GetVersion()
1304 > expectedEventID := startEventID + 1
1305 >
1306 > eventsConsecutiveCheck := func(currentEventId, currentEventVersion int64) error {
1307 > if expectedEventID != currentEventId {
1308 return fmt.Errorf("%w Expected %v, but got %v", ErrEventSlicesNotConsecutive, expectedEventID, currentEventId)
1309 }
1310 > version, err := versionhistory.GetVersionHistoryEventVersion(sourceVersionHistory, currentEventId) workflow_state_replicator.go
1311 > if err != nil {
1312 return serviceerror.NewInternalf("Failed to get version for event id %v from history %v", currentEventId, sourceVersionHistory)
1313 }
1314 > if version != currentEventVersion { workflow_state_replicator.go
1315 return serviceerror.NewInternalf("Event Version does not match. Expected %v, but got %v", version, currentEventVersion)
1316 }
1317 > expectedEventID = currentEventId + 1 workflow_state_replicator.go
1318 > return nil
1319 }
1320 > var historyEvents [][]*historypb.HistoryEvent workflow_state_replicator.go
1321 > for _, blob := range eventBlobs {
1322 > events, err := r.historySerializer.DeserializeEvents(blob)
1323 > if err != nil {
1324 return newBranchToken, err
1325 }
1326 > historyEvents = append(historyEvents, events) workflow_state_replicator.go
1327 }
1328
1329 > ns, err := r.namespaceRegistry.GetNamespaceByID(namespaceID) workflow_state_replicator.go
1330 > if err != nil {
1331 return newBranchToken, err
1332 }
1333 > nsName = ns.Name().String() workflow_state_replicator.go
1334 > // In standby cluster, use a background low priority which is higher than the standby task processing
1335 > callerType := headers.CallerTypeBackgroundLow
1336 > if ns.ActiveClusterName(namespace.RoutingKey{ID: workflowID}) == r.clusterMetadata.GetCurrentClusterName() {
1337 // In active cluster, use lowest priority to minimize the impact to live traffic
1338 callerType = headers.CallerTypePreemptable
1339 }
1340 > quotaRequest := quotas.NewRequest( workflow_state_replicator.go
1341 > "AppendRawHistoryNodes",
1342 > 1,
1343 > nsName,
1344 > callerType,
1345 > 0,
1346 > "",
1347 > )
1348 >
1349 > prevTxnID := localMutableState.GetExecutionInfo().LastFirstEventTxnId
1350 > fetchFromRemoteAndAppend := func(
1351 > startID, // exclusive
1352 > startVersion,
1353 > endID, // exclusive
1354 > endVersion int64) error {
1355 remoteHistoryIterator := collection.NewPagingIterator(r.getHistoryFromRemotePaginationFn(
1356 ctx,
1428 }
1429 // Fill the gap between local last event and request's first event
1430 > if len(historyEvents) > 0 && historyEvents[0][0].EventId > startEventID+1 { workflow_state_replicator.go
1431 err := fetchFromRemoteAndAppend(localLastItem.EventId, localLastItem.Version, historyEvents[0][0].EventId, historyEvents[0][0].Version)
1432 if err != nil {
1441
1442 // add events from request
1443 > for i, events := range historyEvents { workflow_state_replicator.go
1444 > if events[0].EventId <= startEventID {
1445 continue
1446 }
1447 > txnID, err := r.shardContext.GenerateTaskID() workflow_state_replicator.go
1448 > if err != nil {
1449 return newBranchToken, err
1450 }
1451 > for _, event := range events { workflow_state_replicator.go
1452 > err := eventsConsecutiveCheck(event.EventId, event.Version)
1453 > if err != nil {
1454 return newBranchToken, err
1455 }
1456 > localMutableState.AddReapplyCandidateEvent(event) workflow_state_replicator.go
1457 > r.addEventToCache(localMutableState.GetWorkflowKey(), event)
1458 }
1459 > if r.enablePersistenceRateLimiter(nsName) { workflow_state_replicator.go
1460 if err := r.persistenceRateLimiter.Wait(ctx, quotaRequest); err != nil {
1461 return newBranchToken, err
1462 }
1463 }
1464 > _, err = r.executionMgr.AppendRawHistoryNodes(ctx, &persistence.AppendRawHistoryNodesRequest{ workflow_state_replicator.go
1465 > ShardID: r.shardContext.GetShardID(),
1466 > IsNewBranch: isNewBranch,
1467 > BranchToken: versionHistoryToAppend.BranchToken,
1468 > History: eventBlobs[i],
1469 > PrevTransactionID: prevTxnID,
1470 > TransactionID: txnID,
1471 > NodeID: events[0].EventId,
1472 > Info: persistence.BuildHistoryGarbageCleanupInfo(
1473 > namespaceID.String(),
1474 > workflowID,
1475 > runID,
1476 > ),
1477 > })
1478 > if err != nil {
1479 return newBranchToken, err
1480 }
1481 > prevTxnID = txnID workflow_state_replicator.go
1482 > isNewBranch = false
1483 > startEventID = events[len(events)-1].EventId
1484 > startEventVersion = events[len(events)-1].Version
1485 > localMutableState.GetExecutionInfo().ExecutionStats.HistorySize += int64(len(eventBlobs[i].Data))
1486 > if r.shardContext.GetConfig().ExternalPayloadsEnabled(localMutableState.GetNamespaceEntry().Name().String()) {
1487 > externalPayloadSize, externalPayloadCount, err := workflow.CalculateExternalPayloadSize(
1488 > events,
1489 > metrics.NoopMetricsHandler, // don't record metrics since those are not new uploads
1490 > )
1491 > if err != nil {
1492 return newBranchToken, err
1493 }
1494 > localMutableState.AddExternalPayloadSize(externalPayloadSize) workflow_state_replicator.go
1495 > localMutableState.AddExternalPayloadCount(externalPayloadCount)
1496 }
1497 }
1498 // add more events if there is any
1499 > if startEventID < endEventID { workflow_state_replicator.go
1500 err = fetchFromRemoteAndAppend(startEventID, startEventVersion, endEventID+1, endEventVersion)
1501 if err != nil {
2034 workflowKey definition.WorkflowKey,
2035 event *historypb.HistoryEvent,
2036 > ) { workflow_state_replicator.go
2037 > switch event.EventType {
2038 case enumspb.EVENT_TYPE_ACTIVITY_TASK_SCHEDULED,
2039 enumspb.EVENT_TYPE_REQUEST_CANCEL_EXTERNAL_WORKFLOW_EXECUTION_INITIATED,
go.temporal.io/server/service/history/interfaces/shard_context_mock.go 15 introduced LOC · 3 ranges

Open complete file

240
241 // GenerateTaskID mocks base method.
242 > func (m *MockShardContext) GenerateTaskID() (int64, error) { shard_context_mock.go
243 > m.ctrl.T.Helper()
244 > ret := m.ctrl.Call(m, "GenerateTaskID")
245 > ret0, _ := ret[0].(int64)
246 > ret1, _ := ret[1].(error)
247 > return ret0, ret1
248 > }
249
250 // GenerateTaskID indicates an expected call of GenerateTaskID.
251 > func (mr *MockShardContextMockRecorder) GenerateTaskID() *gomock.Call { shard_context_mock.go
252 > mr.mock.ctrl.T.Helper()
253 > return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GenerateTaskID", reflect.TypeOf((*MockShardContext)(nil).GenerateTaskID))
254 > }
255
256 // GenerateTaskIDs mocks base method.
608
609 // GetRemoteAdminClient indicates an expected call of GetRemoteAdminClient.
610 > func (mr *MockShardContextMockRecorder) GetRemoteAdminClient(arg0 any) *gomock.Call { shard_context_mock.go
611 > mr.mock.ctrl.T.Helper()
612 > return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetRemoteAdminClient", reflect.TypeOf((*MockShardContext)(nil).GetRemoteAdminClient), arg0)
613 > }
614
615 // GetReplicationStatus mocks base method.
go.temporal.io/server/service/history/interfaces/mutable_state_mock.go 8 introduced LOC · 2 ranges

Open complete file

411
412 // AddReapplyCandidateEvent mocks base method.
413 > func (m *MockMutableState) AddReapplyCandidateEvent(event *history.HistoryEvent) { mutable_state_mock.go
414 > m.ctrl.T.Helper()
415 > m.ctrl.Call(m, "AddReapplyCandidateEvent", event)
416 > }
417
418 // AddReapplyCandidateEvent indicates an expected call of AddReapplyCandidateEvent.
419 > func (mr *MockMutableStateMockRecorder) AddReapplyCandidateEvent(event any) *gomock.Call { mutable_state_mock.go
420 > mr.mock.ctrl.T.Helper()
421 > return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AddReapplyCandidateEvent", reflect.TypeOf((*MockMutableState)(nil).AddReapplyCandidateEvent), event)
422 > }
423
424 // AddRecordMarkerEvent mocks base method.

Direct hierarchy

Parents

Children