From 6602306557afe5635d056f5a249fe8d7e21a44b9 Mon Sep 17 00:00:00 2001 From: stuart-wells Date: Mon, 26 Oct 2026 13:59:38 -0700 Subject: [PATCH] Require contiguous history for backfillHistory --- .../history/ndc/workflow_state_replicator.go | 30 +++- .../ndc/workflow_state_replicator_test.go | 151 +++++++++++++++++- 2 files changed, 173 insertions(+), 8 deletions(-) diff --git a/service/history/ndc/workflow_state_replicator.go b/service/history/ndc/workflow_state_replicator.go index 07a17888add..007cb0a96d9 100644 --- a/service/history/ndc/workflow_state_replicator.go +++ b/service/history/ndc/workflow_state_replicator.go @@ -2083,6 +2083,7 @@ func (r *WorkflowStateReplicatorImpl) backfillHistory( sortedAncestors := sortAncestors(historyBranch.GetAncestors()) sortedAncestorsIdx := 0 var ancestors []*persistencespb.HistoryBranchRange + expectedEventID := common.FirstEventID BackfillLoop: for remoteHistoryIterator.HasNext() { @@ -2091,14 +2092,29 @@ BackfillLoop: return err } - if isStateBased { - // If backfill suceeds but later event reapply fails, during task's next retry, - // we still need to reapply events that have been stored in local DB. - events, err := r.historySerializer.DeserializeEvents(historyBlob.rawHistory) + // The remote read can come back short or with a gap without returning an error, for example when the + // source moves the branch to another storage tier between pages. Persisting mutable state on top of such + // a history would lose events permanently, so every event is checked and the task retries instead. + batch, err := r.historySerializer.DeserializeEvents(historyBlob.rawHistory) + if err != nil { + return err + } + for _, event := range batch { + if event.GetEventId() != expectedEventID { + return serviceerror.NewInternalf("Backfill history is not contiguous. Expected event %v, but got %v", expectedEventID, event.GetEventId()) + } + version, err := versionhistory.GetVersionHistoryEventVersion(currentVersionHistory, event.GetEventId()) if err != nil { return err } - for _, event := range events { + if version != event.GetVersion() { + return serviceerror.NewInternalf("Backfill event version does not match. Expected %v, but got %v", version, event.GetVersion()) + } + expectedEventID++ + + if isStateBased { + // If backfill suceeds but later event reapply fails, during task's next retry, + // we still need to reapply events that have been stored in local DB. mutableState.AddReapplyCandidateEvent(event) r.addEventToCache(mutableState.GetWorkflowKey(), event) } @@ -2184,6 +2200,10 @@ BackfillLoop: prevBranchID = branchID } + if expectedEventID != lastEventItem.GetEventId()+1 { + return serviceerror.NewInternalf("Backfill history ended early. Expected events up to %v, but got up to %v", lastEventItem.GetEventId(), expectedEventID-1) + } + mutableState.GetExecutionInfo().LastFirstEventTxnId = prevTxnID return mutableState.SetCurrentBranchToken(backfillBranchToken) } diff --git a/service/history/ndc/workflow_state_replicator_test.go b/service/history/ndc/workflow_state_replicator_test.go index 842675bf863..5e0a511539a 100644 --- a/service/history/ndc/workflow_state_replicator_test.go +++ b/service/history/ndc/workflow_state_replicator_test.go @@ -2,6 +2,7 @@ package ndc import ( "context" + "errors" "fmt" "testing" "time" @@ -166,7 +167,7 @@ func (s *workflowReplicatorSuite) Test_ApplyWorkflowState_BrandNew() { BranchToken: historyBranch.GetData(), Items: []*historyspb.VersionHistoryItem{ { - EventId: int64(100), + EventId: int64(6), Version: int64(100), }, }, @@ -218,11 +219,26 @@ func (s *workflowReplicatorSuite) Test_ApplyWorkflowState_BrandNew() { int64(100), ), nil).AnyTimes() s.mockNamespaceCache.EXPECT().GetNamespaceName(namespace.ID(namespaceID)).Return(namespace.Name(namespaceName), nil).AnyTimes() + var historyBlobs []*commonpb.DataBlob + var nodeIDs []int64 + for _, batch := range [][]*historypb.HistoryEvent{ + {{EventId: 1, Version: 100}, {EventId: 2, Version: 100}}, + {{EventId: 3, Version: 100}, {EventId: 4, Version: 100}, {EventId: 5, Version: 100}, {EventId: 6, Version: 100}}, + } { + blob, err := s.serializer.SerializeEvents(batch) + s.NoError(err) + historyBlobs = append(historyBlobs, blob) + nodeIDs = append(nodeIDs, batch[0].GetEventId()) + } s.mockRemoteAdminClient.EXPECT().GetWorkflowExecutionRawHistoryV2(gomock.Any(), gomock.Any()).Return( - &adminservice.GetWorkflowExecutionRawHistoryV2Response{}, + &adminservice.GetWorkflowExecutionRawHistoryV2Response{ + HistoryBatches: historyBlobs, + HistoryNodeIds: nodeIDs, + }, nil, ) s.mockExecutionManager.EXPECT().ReadHistoryBranchByBatch(gomock.Any(), gomock.Any()).Return(nil, serviceerror.NewNotFound("test")) + s.mockExecutionManager.EXPECT().AppendRawHistoryNodes(gomock.Any(), gomock.Any()).Return(nil, nil).Times(2) s.mockExecutionManager.EXPECT().GetCurrentExecution(gomock.Any(), gomock.Any()).Return(nil, serviceerror.NewNotFound("")) fakeStartHistory := &historypb.HistoryEvent{ Attributes: &historypb.HistoryEvent_WorkflowExecutionStartedEventAttributes{ @@ -275,7 +291,7 @@ func (s *workflowReplicatorSuite) Test_ApplyWorkflowState_Ancestors() { BranchToken: historyBranch.GetData(), Items: []*historyspb.VersionHistoryItem{ { - EventId: int64(100), + EventId: int64(6), Version: int64(100), }, }, @@ -342,9 +358,11 @@ func (s *workflowReplicatorSuite) Test_ApplyWorkflowState_Ancestors() { Events: []*historypb.HistoryEvent{ { EventId: 1, + Version: 100, }, { EventId: 2, + Version: 100, }, }, }, @@ -352,6 +370,7 @@ func (s *workflowReplicatorSuite) Test_ApplyWorkflowState_Ancestors() { Events: []*historypb.HistoryEvent{ { EventId: 3, + Version: 100, }, }, }, @@ -359,6 +378,7 @@ func (s *workflowReplicatorSuite) Test_ApplyWorkflowState_Ancestors() { Events: []*historypb.HistoryEvent{ { EventId: 4, + Version: 100, }, }, }, @@ -366,9 +386,11 @@ func (s *workflowReplicatorSuite) Test_ApplyWorkflowState_Ancestors() { Events: []*historypb.HistoryEvent{ { EventId: 5, + Version: 100, }, { EventId: 6, + Version: 100, }, }, }, @@ -420,6 +442,129 @@ func (s *workflowReplicatorSuite) Test_ApplyWorkflowState_Ancestors() { s.NoError(err) } +func (s *workflowReplicatorSuite) TestApplyWorkflowStateBackfillIncomplete() { + testCases := []struct { + name string + remoteBatches [][]*historypb.HistoryEvent + appends int + errorContains string + }{ + { + // The source returned the ancestor range but nothing from the run's own branch. + name: "ends at ancestor", + remoteBatches: [][]*historypb.HistoryEvent{ + {{EventId: 1, Version: 100}, {EventId: 2, Version: 100}}, + {{EventId: 3, Version: 100}}, + }, + appends: 2, + errorContains: "Backfill history ended early. Expected events up to 6, but got up to 3", + }, + { + name: "gap", + remoteBatches: [][]*historypb.HistoryEvent{ + {{EventId: 1, Version: 100}, {EventId: 2, Version: 100}}, + {{EventId: 5, Version: 100}, {EventId: 6, Version: 100}}, + }, + appends: 1, + errorContains: "Backfill history is not contiguous. Expected event 3, but got 5", + }, + { + name: "version mismatch", + remoteBatches: [][]*historypb.HistoryEvent{ + {{EventId: 1, Version: 100}, {EventId: 2, Version: 100}}, + {{EventId: 3, Version: 99}}, + }, + appends: 1, + errorContains: "Backfill event version does not match. Expected 100, but got 99", + }, + } + + for _, tc := range testCases { + s.Run(tc.name, func() { + namespaceID := uuid.NewString() + runID := uuid.NewString() + branchInfo := &persistencespb.HistoryBranch{ + TreeId: runID, + BranchId: uuid.NewString(), + Ancestors: []*persistencespb.HistoryBranchRange{ + { + BranchId: uuid.NewString(), + BeginNodeId: 1, + EndNodeId: 4, + }, + }, + } + historyBranch, err := s.serializer.HistoryBranchToBlob(branchInfo) + s.NoError(err) + request := &historyservice.ReplicateWorkflowStateRequest{ + WorkflowState: &persistencespb.WorkflowMutableState{ + ExecutionInfo: &persistencespb.WorkflowExecutionInfo{ + WorkflowId: s.workflowID, + NamespaceId: namespaceID, + VersionHistories: &historyspb.VersionHistories{ + Histories: []*historyspb.VersionHistory{ + { + BranchToken: historyBranch.GetData(), + Items: []*historyspb.VersionHistoryItem{{EventId: 6, Version: 100}}, + }, + }, + }, + CompletionEventBatchId: 5, + }, + ExecutionState: &persistencespb.WorkflowExecutionState{ + RunId: runID, + State: enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED, + Status: enumspb.WORKFLOW_EXECUTION_STATUS_TERMINATED, + }, + NextEventId: 7, + }, + RemoteCluster: "test", + NamespaceId: namespaceID, + } + mockWeCtx := historyi.NewMockWorkflowContext(s.controller) + s.mockWorkflowCache.EXPECT().GetOrCreateWorkflowExecution( + gomock.Any(), + s.mockShard, + namespace.ID(namespaceID), + &commonpb.WorkflowExecution{WorkflowId: s.workflowID, RunId: runID}, + locks.PriorityHigh, + ).Return(mockWeCtx, wcache.NoopReleaseFn, nil) + mockWeCtx.EXPECT().LoadMutableState(gomock.Any(), s.mockShard).Return(nil, serviceerror.NewNotFound("ms not found")) + s.mockNamespaceCache.EXPECT().GetNamespaceByID(namespace.ID(namespaceID)).Return(namespace.NewNamespaceForTest( + &persistencespb.NamespaceInfo{Name: "namespaceName"}, + nil, + false, + nil, + int64(100), + ), nil).AnyTimes() + + var historyBlobs []*commonpb.DataBlob + var nodeIDs []int64 + for _, batch := range tc.remoteBatches { + blob, err := s.serializer.SerializeEvents(batch) + s.NoError(err) + historyBlobs = append(historyBlobs, blob) + nodeIDs = append(nodeIDs, batch[0].GetEventId()) + } + s.mockRemoteAdminClient.EXPECT().GetWorkflowExecutionRawHistoryV2(gomock.Any(), gomock.Any()).Return( + &adminservice.GetWorkflowExecutionRawHistoryV2Response{ + HistoryBatches: historyBlobs, + HistoryNodeIds: nodeIDs, + }, + nil, + ) + s.mockExecutionManager.EXPECT().ReadHistoryBranchByBatch(gomock.Any(), gomock.Any()).Return(nil, serviceerror.NewNotFound("test")) + s.mockExecutionManager.EXPECT().AppendRawHistoryNodes(gomock.Any(), gomock.Any()).Return(nil, nil).Times(tc.appends) + + err = s.workflowStateReplicator.SyncWorkflowState(context.Background(), request) + s.ErrorContains(err, tc.errorContains) + // Internal errors are retried by replication tasks; InvalidArgument and DataLoss are not. + _, isInternal := errors.AsType[*serviceerror.Internal](err) + s.True(isInternal) + }) + } +} + func (s *workflowReplicatorSuite) Test_ApplyWorkflowState_NoClosedWorkflow_Error() { err := s.workflowStateReplicator.SyncWorkflowState(context.Background(), &historyservice.ReplicateWorkflowStateRequest{ WorkflowState: &persistencespb.WorkflowMutableState{