Add refKey to taks and rename nodeKey in reference

e2e/orchestrator_event_map_task_map_task_test.go

@@ -61,8 +61,8 @@ func testOrchestratorEventMapTaskMapTask(executionStream pb.Execution_StreamClie
 									"msg": {
 										Value: &process.Process_Node_Map_Output_Ref{
 											Ref: &process.Process_Node_Map_Output_Reference{
-												NodeKey: "n0",
-												Key:     "msg",
+												RefKey: "n0",
+												Key:    "msg",
 											},
 										},
 									},

e2e/orchestrator_result_map_task_map_task_test.go

@@ -60,8 +60,8 @@ func testOrchestratorResultMapTaskMapTask(executionStream pb.Execution_StreamCli
 									"msg": {
 										Value: &process.Process_Node_Map_Output_Ref{
 											Ref: &process.Process_Node_Map_Output_Reference{
-												NodeKey: "n0",
-												Key:     "msg",
+												RefKey: "n0",
+												Key:    "msg",
 											},
 										},
 									},

orchestrator/orchestrator.go

@@ -202,14 +202,14 @@ func (s *Orchestrator) outputToValue(output *process.Process_Node_Map_Output, wf
 			},
 		}, nil
 	case *process.Process_Node_Map_Output_Ref:
-		node, err := wf.FindNode(v.Ref.NodeKey)
-		if err != nil {
-			return nil, err
-		}
-		if node.GetTask() != nil {
-			return s.resolveInput(wf.Hash, exec, v.Ref.NodeKey, v.Ref.Key)
+		nodes := wf.FindNodes(func(n *process.Process_Node) bool {
+			task := n.GetTask()
+			return task != nil && task.RefKey == v.Ref.RefKey
+		})
+		if len(nodes) != 1 {
+			return nil, fmt.Errorf("reference's key not found")
 		}
-		return data.Fields[v.Ref.Key], nil
+		return s.resolveInput(wf.Hash, exec, v.Ref.RefKey, v.Ref.Key)
 	default:
 		return nil, errors.New("unknown output")
 	}

process/process.go

@@ -22,7 +22,7 @@ func (w *Process) Validate() error {
 		if mapNode != nil {
 			for _, output := range mapNode.Outputs {
 				if ref := output.GetRef(); ref != nil {
-					if _, err := w.FindNode(ref.NodeKey); err != nil {
+					if _, err := w.FindNode(ref.RefKey); err != nil {
 						return err
 					}
 				}