Handle data loss error in replication read history (#3704)

* Bypass data corruption error in replication

Author: yux0

common/dynamicconfig/constants.go

@@ -663,6 +663,8 @@ const (
 	ReplicationTaskProcessorHostQPS = "history.ReplicationTaskProcessorHostQPS"
 	// ReplicationTaskProcessorShardQPS is the qps of task processing rate limiter on shard level
 	ReplicationTaskProcessorShardQPS = "history.ReplicationTaskProcessorShardQPS"
+	// ReplicationBypassCorruptedData is the flag to bypass corrupted workflow data in source cluster
+	ReplicationBypassCorruptedData = "history.ReplicationBypassCorruptedData"
 
 	// keys for worker
 

common/persistence/history_manager.go

@@ -33,6 +33,7 @@ import (
 	enumspb "go.temporal.io/api/enums/v1"
 	historypb "go.temporal.io/api/history/v1"
 	"go.temporal.io/api/serviceerror"
+
 	"go.temporal.io/server/common/persistence/serialization"
 
 	persistencespb "go.temporal.io/server/api/persistence/v1"
@@ -1159,9 +1160,9 @@ func (m *executionManagerImpl) filterHistoryNodes(
 
 		switch {
 		case node.NodeID < lastNodeID:
-			return nil, serviceerror.NewUnavailable("corrupted data, nodeID cannot decrease")
+			return nil, serviceerror.NewDataLoss("corrupted data, nodeID cannot decrease")
 		case node.NodeID == lastNodeID:
-			return nil, serviceerror.NewUnavailable("corrupted data, same nodeID must have smaller txnID")
+			return nil, serviceerror.NewDataLoss("corrupted data, same nodeID must have smaller txnID")
 		default: // row.NodeID > lastNodeID:
 			// NOTE: when row.nodeID > lastNodeID, we expect the one with largest txnID comes first
 			lastTransactionID = node.TransactionID
@@ -1188,7 +1189,7 @@ func (m *executionManagerImpl) filterHistoryNodesReverse(
 
 		switch {
 		case node.NodeID > lastNodeID:
-			return nil, serviceerror.NewUnavailable("corrupted data, nodeID cannot decrease")
+			return nil, serviceerror.NewDataLoss("corrupted data, nodeID cannot decrease")
 		default:
 			lastTransactionID = node.PrevTransactionID
 			lastNodeID = node.NodeID

service/history/configs/config.go

@@ -241,6 +241,7 @@ type Config struct {
 	ReplicationTaskProcessorCleanupJitterCoefficient     dynamicconfig.FloatPropertyFnWithShardIDFilter
 	ReplicationTaskProcessorHostQPS                      dynamicconfig.FloatPropertyFn
 	ReplicationTaskProcessorShardQPS                     dynamicconfig.FloatPropertyFn
+	ReplicationBypassCorruptedData                       dynamicconfig.BoolPropertyFnWithNamespaceIDFilter
 
 	// The following are used by consistent query
 	MaxBufferedQueryCount dynamicconfig.IntPropertyFn
@@ -422,6 +423,7 @@ func NewConfig(dc *dynamicconfig.Collection, numberOfShards int32, isAdvancedVis
 		ReplicatorProcessorMaxSkipTaskCount:                   dc.GetIntProperty(dynamicconfig.ReplicatorMaxSkipTaskCount, 250),
 		ReplicationTaskProcessorHostQPS:                       dc.GetFloat64Property(dynamicconfig.ReplicationTaskProcessorHostQPS, 1500),
 		ReplicationTaskProcessorShardQPS:                      dc.GetFloat64Property(dynamicconfig.ReplicationTaskProcessorShardQPS, 30),
+		ReplicationBypassCorruptedData:                        dc.GetBoolPropertyFnWithNamespaceIDFilter(dynamicconfig.ReplicationBypassCorruptedData, false),
 
 		MaximumBufferedEventsBatch:      dc.GetIntProperty(dynamicconfig.MaximumBufferedEventsBatch, 100),
 		MaximumSignalsPerExecution:      dc.GetIntPropertyFilteredByNamespace(dynamicconfig.MaximumSignalsPerExecution, 0),

service/history/replication/ack_manager.go

@@ -485,7 +485,7 @@ func (p *ackMgrImpl) generateHistoryReplicationTask(
 				taskInfo.NextEventID,
 			)
 			if err != nil {
-				return nil, err
+				return nil, p.handleReadHistoryError(namespaceID, workflowID, runID, err)
 			}
 
 			replicationTask := &replicationspb.ReplicationTask{
@@ -689,7 +689,7 @@ func (p *ackMgrImpl) processNewRunReplication(
 			common.FirstEventID+1,
 		)
 		if err != nil {
-			return nil, err
+			return nil, p.handleReadHistoryError(namespaceID, workflowID, newRunID, err)
 		}
 	}
 	attr.HistoryTaskAttributes.NewRunEvents = newRunEventsBlob
@@ -720,3 +720,26 @@ func getVersionHistoryItems(
 	}
 	return versionhistory.CopyVersionHistory(versionHistoryBranch).GetItems(), versionHistoryBranch.GetBranchToken(), nil
 }
+
+func (p *ackMgrImpl) handleReadHistoryError(
+	namespaceID namespace.ID,
+	workflowID string,
+	runID string,
+	err error,
+) error {
+	switch err.(type) {
+	case *serviceerror.NotFound, *serviceerror.DataLoss:
+		if p.config.ReplicationBypassCorruptedData(namespaceID.String()) {
+			// bypass this corrupted workflow to unblock the replication queue.
+			p.logger.Error("Cannot get history from corrupted workflow",
+				tag.WorkflowNamespaceID(namespaceID.String()),
+				tag.WorkflowID(workflowID),
+				tag.WorkflowRunID(runID),
+				tag.Error(err))
+			return nil
+		}
+		return err
+	default:
+		return err
+	}
+}