Retry getHistory request from passive cluster in case of replication lag (#952)

* Retry getHistory request from passive cluster in case of replication lag
* Fix unit test

Author: meiliang86

internal/internal_workflow_client.go

@@ -487,11 +487,16 @@ func (wc *workflowClient) TerminateWorkflow(ctx context.Context, workflowID stri
 }
 
 // GetWorkflowHistory return a channel which contains the history events of a given workflow
-func (wc *workflowClient) GetWorkflowHistory(ctx context.Context, workflowID string, runID string,
-	isLongPoll bool, filterType s.HistoryEventFilterType) HistoryEventIterator {
+func (wc *workflowClient) GetWorkflowHistory(
+	ctx context.Context,
+	workflowID string,
+	runID string,
+	isLongPoll bool,
+	filterType s.HistoryEventFilterType,
+) HistoryEventIterator {
 
 	domain := wc.domain
-	paginate := func(nexttoken []byte) (*s.GetWorkflowExecutionHistoryResponse, error) {
+	paginate := func(nextToken []byte) (*s.GetWorkflowExecutionHistoryResponse, error) {
 		request := &s.GetWorkflowExecutionHistoryRequest{
 			Domain: common.StringPtr(domain),
 			Execution: &s.WorkflowExecution{
@@ -500,7 +505,8 @@ func (wc *workflowClient) GetWorkflowHistory(ctx context.Context, workflowID str
 			},
 			WaitForNewEvent:        common.BoolPtr(isLongPoll),
 			HistoryEventFilterType: &filterType,
-			NextPageToken:          nexttoken,
+			NextPageToken:          nextToken,
+			SkipArchival:           common.BoolPtr(isLongPoll),
 		}
 
 		var response *s.GetWorkflowExecutionHistoryResponse
@@ -530,7 +536,12 @@ func (wc *workflowClient) GetWorkflowHistory(ctx context.Context, workflowID str
 						response.History = history
 					}
 					return err1
-				}, createDynamicServiceRetryPolicy(ctx), isServiceTransientError)
+				},
+				createDynamicServiceRetryPolicy(ctx),
+				func(err error) bool {
+					return isServiceTransientError(err) || isEntityNonExistFromPassive(err)
+				},
+			)
 
 			if err != nil {
 				return nil, err
@@ -549,6 +560,16 @@ func (wc *workflowClient) GetWorkflowHistory(ctx context.Context, workflowID str
 	}
 }
 
+func isEntityNonExistFromPassive(err error) bool {
+	if nonExistError, ok := err.(*s.EntityNotExistsError); ok {
+		return nonExistError.GetActiveCluster() != "" &&
+			nonExistError.GetCurrentCluster() != "" &&
+			nonExistError.GetActiveCluster() != nonExistError.GetCurrentCluster()
+	}
+
+	return false
+}
+
 // CompleteActivity reports activity completed. activity Execute method can return activity.ErrResultPending to
 // indicate the activity is not completed when it's Execute method returns. In that case, this CompleteActivity() method
 // should be called when that activity is completed with the actual result and error. If err is nil, activity task

internal/internal_workflow_client_test.go

@@ -934,6 +934,7 @@ func getGetWorkflowExecutionHistoryRequest(filterType shared.HistoryEventFilterT
 		},
 		WaitForNewEvent:        common.BoolPtr(isLongPoll),
 		HistoryEventFilterType: &filterType,
+		SkipArchival: common.BoolPtr(true),
 	}
 
 	return request