Fail force-replication early if any GenerateReplicationTasks/Generate…

…AndVerifyReplicationTasks activity returns error
temporalio · Jul 18, 2023 · ca5bf4e · ca5bf4e
1 parent 45d340d
commit ca5bf4e
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 2 deletions.
diff --git a/service/worker/migration/activities.go b/service/worker/migration/activities.go
@@ -645,7 +645,7 @@ func (a *activities) verifyReplicationTasks(
 			return false, progress, nil
 
 		default:
-			return false, progress, err
+			return false, progress, fmt.Errorf("remoteClient.DescribeMutableState failed. error: %v", err)
 		}
 	}
 

diff --git a/service/worker/migration/force_replication_workflow.go b/service/worker/migration/force_replication_workflow.go
@@ -371,6 +371,7 @@ func enqueueReplicationTasks(ctx workflow.Context, workflowExecutionsCh workflow
 	var a *activities
 	var futures []workflow.Future
 	var workflowExecutions []commonpb.WorkflowExecution
+	var lastActivityErr error
 
 	for workflowExecutionsCh.Receive(ctx, &workflowExecutions) {
 		var replicationTaskFuture workflow.Future
@@ -394,10 +395,17 @@ func enqueueReplicationTasks(ctx workflow.Context, workflowExecutionsCh workflow
 		pendingActivities++
 		selector.AddFuture(replicationTaskFuture, func(f workflow.Future) {
 			pendingActivities--
+
+			if err := f.Get(ctx, nil); err != nil {
+				lastActivityErr = err
+			}
 		})
 
-		if pendingActivities == params.ConcurrentActivityCount {
+		if pendingActivities >= params.ConcurrentActivityCount {
 			selector.Select(ctx) // this will block until one of the in-flight activities completes
+			if lastActivityErr != nil {
+				return lastActivityErr
+			}
 		}
 
 		futures = append(futures, replicationTaskFuture)