Add error handling to session setup

flow/workflows/cdc_flow.go

@@ -378,19 +378,34 @@ func CDCFlowWorkflowWithConfig(
 		HeartbeatTimeout:    time.Minute,
 		WaitForCancellation: true,
 	})
-	workflow.ExecuteActivity(
+	fMaintain := workflow.ExecuteActivity(
 		syncCtx,
 		flowable.MaintainPull,
 		cfg,
 		sessionInfo.SessionID,
 	)
-	fWait := workflow.ExecuteActivity(
+	fSessionSetup := workflow.ExecuteActivity(
 		syncCtx,
 		flowable.WaitForSourceConnector,
 		sessionInfo.SessionID,
 	)
-	if err := fWait.Get(syncCtx, nil); err != nil {
-		return nil, err
+
+	var sessionError error
+	sessionSelector := workflow.NewNamedSelector(ctx, "Session Setup")
+	sessionSelector.AddFuture(fMaintain, func(f workflow.Future) {
+		// MaintainPull should never exit without an error before this point
+		sessionError = f.Get(syncCtx, nil)
+	})
+	sessionSelector.AddFuture(fSessionSetup, func(f workflow.Future) {
+		// Happy path is waiting for this to return without error
+		sessionError = f.Get(syncCtx, nil)
+	})
+	sessionSelector.AddReceive(ctx.Done(), func(_ workflow.ReceiveChannel, _ bool) {
+		sessionError = ctx.Err()
+	})
+	sessionSelector.Select(ctx)
+	if sessionError != nil {
+		return nil, sessionError
 	}
 
 	state.SyncFlowOptions = &protos.SyncFlowOptions{
@@ -452,7 +467,7 @@ func CDCFlowWorkflowWithConfig(
 
 	var canceled bool
 	flowSignalChan := workflow.GetSignalChannel(ctx, shared.FlowSignalName)
-	mainLoopSelector := workflow.NewSelector(ctx)
+	mainLoopSelector := workflow.NewNamedSelector(ctx, "Main Loop")
 	mainLoopSelector.AddReceive(ctx.Done(), func(_ workflow.ReceiveChannel, _ bool) {
 		canceled = true
 	})