Vihren Atlas

workflow_state_replicator.go ×13

Code frontier

unlabeled · c_d1561fd90fba

3 tests · 4771 LOC · 201 files · introduces 0 tests · 66 LOC · 2 files

Introduces — evidence that enters the hierarchy at this concept

Code
14 ranges66 lines · 2 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
767 ranges4771 lines · 201 files · Browse complete extent
All tests (intent)
3 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.

2 files ranked by introduced lines: 66 introduced LOC across 14 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/service/history/ndc/workflow_state_replicator.go 59 introduced LOC · 13 ranges

Open complete file

1353 endID, // exclusive
1354 endVersion int64) error {
1355 > remoteHistoryIterator := collection.NewPagingIterator(r.getHistoryFromRemotePaginationFn( workflow_state_replicator.go
1356 > ctx,
1357 > sourceClusterName,
1358 > namespaceID,
1359 > workflowID,
1360 > runID,
1361 > startID,
1362 > startVersion,
1363 > endID,
1364 > endVersion),
1365 > )
1366 > for remoteHistoryIterator.HasNext() {
1367 > historyBlob, err := remoteHistoryIterator.Next()
1368 > if err != nil {
1369 return err
1370 }
1371
1372 > txnID, err := r.shardContext.GenerateTaskID() workflow_state_replicator.go
1373 > if err != nil {
1374 return err
1375 }
1376 > events, err := r.historySerializer.DeserializeEvents(historyBlob.rawHistory) workflow_state_replicator.go
1377 > if err != nil {
1378 return err
1379 }
1380 > for _, event := range events { workflow_state_replicator.go
1381 > err := eventsConsecutiveCheck(event.EventId, event.Version)
1382 > if err != nil {
1383 return err
1384 }
1385 > localMutableState.AddReapplyCandidateEvent(event) workflow_state_replicator.go
1386 > r.addEventToCache(localMutableState.GetWorkflowKey(), event)
1387 }
1388 > if r.enablePersistenceRateLimiter(nsName) { workflow_state_replicator.go
1389 if err := r.persistenceRateLimiter.Wait(ctx, quotaRequest); err != nil {
1390 return err
1392 }
1393
1394 > _, err = r.executionMgr.AppendRawHistoryNodes(ctx, &persistence.AppendRawHistoryNodesRequest{ workflow_state_replicator.go
1395 > ShardID: r.shardContext.GetShardID(),
1396 > IsNewBranch: isNewBranch,
1397 > BranchToken: versionHistoryToAppend.BranchToken,
1398 > History: historyBlob.rawHistory,
1399 > PrevTransactionID: prevTxnID,
1400 > TransactionID: txnID,
1401 > NodeID: historyBlob.nodeID,
1402 > Info: persistence.BuildHistoryGarbageCleanupInfo(
1403 > namespaceID.String(),
1404 > workflowID,
1405 > runID,
1406 > ),
1407 > })
1408 > if err != nil {
1409 return err
1410 }
1411 > prevTxnID = txnID workflow_state_replicator.go
1412 > isNewBranch = false
1413 >
1414 > localMutableState.GetExecutionInfo().ExecutionStats.HistorySize += int64(len(historyBlob.rawHistory.Data))
1415 > if r.shardContext.GetConfig().ExternalPayloadsEnabled(nsName) {
1416 > externalPayloadSize, externalPayloadCount, err := workflow.CalculateExternalPayloadSize(
1417 > events,
1418 > metrics.NoopMetricsHandler, // don't record metrics since those are not new uploads
1419 > )
1420 > if err != nil {
1421 return err
1422 }
1423 > localMutableState.AddExternalPayloadSize(externalPayloadSize) workflow_state_replicator.go
1424 > localMutableState.AddExternalPayloadCount(externalPayloadCount)
1425 }
1426 }
1427 > return nil workflow_state_replicator.go
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 {
1431 > err := fetchFromRemoteAndAppend(localLastItem.EventId, localLastItem.Version, historyEvents[0][0].EventId, historyEvents[0][0].Version) workflow_state_replicator.go
1432 > if err != nil {
1433 return newBranchToken, err
1434 }
1435 > startEventID = historyEvents[0][0].EventId - 1 workflow_state_replicator.go
1436 > startEventVersion, err = versionhistory.GetVersionHistoryEventVersion(sourceVersionHistory, startEventID)
1437 > if err != nil {
1438 return newBranchToken, err
1439 }
1498 // add more events if there is any
1499 if startEventID < endEventID {
1500 > err = fetchFromRemoteAndAppend(startEventID, startEventVersion, endEventID+1, endEventVersion) workflow_state_replicator.go
1501 > if err != nil {
1502 return newBranchToken, err
1503 }
go.temporal.io/server/service/history/interfaces/shard_context_mock.go 7 introduced LOC · 1 range

Open complete file

599
600 // GetRemoteAdminClient mocks base method.
601 > func (m *MockShardContext) GetRemoteAdminClient(arg0 string) (adminservice.AdminServiceClient, error) { shard_context_mock.go
602 > m.ctrl.T.Helper()
603 > ret := m.ctrl.Call(m, "GetRemoteAdminClient", arg0)
604 > ret0, _ := ret[0].(adminservice.AdminServiceClient)
605 > ret1, _ := ret[1].(error)
606 > return ret0, ret1
607 > }
608
609 // GetRemoteAdminClient indicates an expected call of GetRemoteAdminClient.

Direct hierarchy

Parents

Children