Repository: bacalhau-project/bacalhau (language: python)

Issue:
Some inputs fail when writing to the same target
Running a job with multiple inputs, where they all write to the same folder can fail with an obscure error.

```
bacalhau \
    docker run \
    -i src=ipfs://qmbrr4kuxmxqfzpnlusb1ksmdvftbucv1hvsdaauke4epj/,dst=/inputs/data \
    -i src=ipfs://qmbrr4kuxmxqfzpnlusb1ksmdvftbucv1hvsdaauke4epj/,dst=/inputs/data \
ubuntu echo hello
```

fails (in logs/events) with 

```
failed to copy /inputs/data to volume: failed to write to ''/tmp/bacalhau-ipfs900595700/QmbyYag1VXo45qPG9UVfbF6hVe5aHj2pR61voFeuA7pTH8'':
      path already exists and overwriting is not allowed
```

We should investigate whether it is possible to validate the inputs earlier to avoid the users waiting for something that will certainly fail.


Discussion / hints:
still needed?

Fix patch:
diff --git a/pkg/models/input_source.go b/pkg/models/input_source.go
index d253c22d6e..53f6c087d7 100644
--- a/pkg/models/input_source.go
+++ b/pkg/models/input_source.go
@@ -46,6 +46,7 @@ func (a *InputSource) Copy() *InputSource {
 	}
 	return &InputSource{
 		Source: a.Source.Copy(),
+		Alias:  a.Alias,
 		Target: a.Target,
 	}
 }
diff --git a/pkg/models/task.go b/pkg/models/task.go
index ae7fd214d2..4a5706e991 100644
--- a/pkg/models/task.go
+++ b/pkg/models/task.go
@@ -139,12 +139,40 @@ func (t *Task) ValidateSubmission() error {
 		mErr = errors.Join(mErr, errors.New("publisher must be set if result paths are set"))
 	}
 
+	// Check for collisions in input sources
 	seenInputAliases := make(map[string]bool)
+	seenInputTargets := make(map[string]bool)
 	for _, input := range t.InputSources {
-		if input.Alias != "" && seenInputAliases[input.Alias] {
-			mErr = errors.Join(mErr, fmt.Errorf("input source with alias %s already exist", input.Alias))
+		if input.Alias != "" {
+			if seenInputAliases[input.Alias] {
+				mErr = errors.Join(mErr, fmt.Errorf("input source with alias '%s' already exists", input.Alias))
+			}
+			seenInputAliases[input.Alias] = true
+		}
+		if input.Target != "" {
+			if seenInputTargets[input.Target] {
+				mErr = errors.Join(mErr, fmt.Errorf("input source with target '%s' already exists", input.Target))
+			}
+			seenInputTargets[input.Target] = true
+		}
+	}
+
+	// Check for collisions in result paths
+	seenResultNames := make(map[string]bool)
+	seenResultPaths := make(map[string]bool)
+	for _, result := range t.ResultPaths {
+		if result.Name != "" {
+			if seenResultNames[result.Name] {
+				mErr = errors.Join(mErr, fmt.Errorf("result path with name '%s' already exists", result.Name))
+			}
+			seenResultNames[result.Name] = true
+		}
+		if result.Path != "" {
+			if seenResultPaths[result.Path] {
+				mErr = errors.Join(mErr, fmt.Errorf("result path '%s' already exists", result.Path))
+			}
+			seenResultPaths[result.Path] = true
 		}
-		seenInputAliases[input.Alias] = true
 	}
 
 	return mErr
@@ -156,9 +184,14 @@ func (t *Task) ToBuilder() *TaskBuilder {
 }
 
 func (t *Task) AllStorageTypes() []string {
-	var types []string
+	uniqueTypes := make(map[string]bool)
 	for _, a := range t.InputSources {
-		types = append(types, a.Source.Type)
+		uniqueTypes[a.Source.Type] = true
+	}
+
+	types := make([]string, 0, len(uniqueTypes))
+	for typ := range uniqueTypes {
+		types = append(types, typ)
 	}
 	return types
 }


Test patch:
diff --git a/pkg/models/task_test.go b/pkg/models/task_test.go
new file mode 100644
index 0000000000..bbe7716a5e
--- /dev/null
+++ b/pkg/models/task_test.go
@@ -0,0 +1,214 @@
+//go:build unit || !integration
+
+package models
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/suite"
+)
+
+type TaskTestSuite struct {
+	suite.Suite
+}
+
+func TestTaskSuite(t *testing.T) {
+	suite.Run(t, new(TaskTestSuite))
+}
+
+func (suite *TaskTestSuite) TestTaskNormalization() {
+	task := &Task{
+		Name:      "test-task",
+		Engine:    &SpecConfig{Type: "docker"},
+		Publisher: &SpecConfig{Type: "s3"},
+		InputSources: []*InputSource{
+			{Alias: "input1", Target: "/input1", Source: &SpecConfig{Type: "http"}},
+		},
+		ResultPaths: []*ResultPath{
+			{Name: "output1", Path: "/output1"},
+		},
+		Meta:            nil,
+		Env:             nil,
+		ResourcesConfig: nil,
+		Network:         nil,
+		Timeouts:        nil,
+	}
+
+	task.Normalize()
+
+	suite.NotNil(task.Meta)
+	suite.NotNil(task.Env)
+	suite.NotNil(task.ResourcesConfig)
+	suite.NotNil(task.Network)
+	suite.NotNil(task.Timeouts)
+	suite.NotEmpty(task.InputSources)
+	suite.NotEmpty(task.ResultPaths)
+}
+
+func (suite *TaskTestSuite) TestTaskValidation() {
+	tests := []struct {
+		name    string
+		task    *Task
+		wantErr bool
+		errMsg  string
+	}{
+		{
+			name: "Valid task",
+			task: &Task{
+				Name:   "valid-task",
+				Engine: &SpecConfig{Type: "docker"},
+				InputSources: []*InputSource{
+					{Alias: "input1", Target: "/input1", Source: &SpecConfig{Type: "http"}},
+					{Alias: "input2", Target: "/input2", Source: &SpecConfig{Type: "http"}},
+				},
+				ResultPaths: []*ResultPath{
+					{Name: "output1", Path: "/output1"},
+					{Name: "output2", Path: "/output2"},
+				},
+				Publisher: &SpecConfig{Type: "s3"},
+			},
+			wantErr: false,
+		},
+		{
+			name: "Empty task name",
+			task: &Task{
+				Name:   "",
+				Engine: &SpecConfig{Type: "docker"},
+			},
+			wantErr: true,
+			errMsg:  "missing task name",
+		},
+		{
+			name: "Duplicate input source alias",
+			task: &Task{
+				Name:   "duplicate-alias",
+				Engine: &SpecConfig{Type: "docker"},
+				InputSources: []*InputSource{
+					{Alias: "input1", Target: "/input1", Source: &SpecConfig{Type: "http"}},
+					{Alias: "input1", Target: "/input2", Source: &SpecConfig{Type: "http"}},
+				},
+			},
+			wantErr: true,
+			errMsg:  "input source with alias 'input1' already exists",
+		},
+		{
+			name: "Duplicate input source target",
+			task: &Task{
+				Name:   "duplicate-target",
+				Engine: &SpecConfig{Type: "docker"},
+				InputSources: []*InputSource{
+					{Alias: "input1", Target: "/input", Source: &SpecConfig{Type: "http"}},
+					{Alias: "input2", Target: "/input", Source: &SpecConfig{Type: "http"}},
+				},
+			},
+			wantErr: true,
+			errMsg:  "input source with target '/input' already exists",
+		},
+		{
+			name: "Duplicate result path name",
+			task: &Task{
+				Name:   "duplicate-result-name",
+				Engine: &SpecConfig{Type: "docker"},
+				ResultPaths: []*ResultPath{
+					{Name: "output", Path: "/output1"},
+					{Name: "output", Path: "/output2"},
+				},
+				Publisher: &SpecConfig{Type: "s3"},
+			},
+			wantErr: true,
+			errMsg:  "result path with name 'output' already exists",
+		},
+		{
+			name: "Duplicate result path",
+			task: &Task{
+				Name:   "duplicate-result-path",
+				Engine: &SpecConfig{Type: "docker"},
+				ResultPaths: []*ResultPath{
+					{Name: "output1", Path: "/output"},
+					{Name: "output2", Path: "/output"},
+				},
+				Publisher: &SpecConfig{Type: "s3"},
+			},
+			wantErr: true,
+			errMsg:  "result path '/output' already exists",
+		},
+		{
+			name: "Result paths without publisher",
+			task: &Task{
+				Name:   "missing-publisher",
+				Engine: &SpecConfig{Type: "docker"},
+				ResultPaths: []*ResultPath{
+					{Name: "output", Path: "/output"},
+				},
+			},
+			wantErr: true,
+			errMsg:  "publisher must be set if result paths are set",
+		},
+	}
+
+	for _, tt := range tests {
+		suite.Run(tt.name, func() {
+			err := tt.task.ValidateSubmission()
+			if tt.wantErr {
+				suite.Error(err)
+				suite.Contains(err.Error(), tt.errMsg)
+			} else {
+				suite.NoError(err)
+			}
+		})
+	}
+}
+
+func (suite *TaskTestSuite) TestTaskCopy() {
+	original := &Task{
+		Name:      "original-task",
+		Engine:    &SpecConfig{Type: "docker"},
+		Publisher: &SpecConfig{Type: "s3"},
+		InputSources: []*InputSource{
+			{Alias: "input1", Target: "/input1", Source: &SpecConfig{Type: "http"}},
+		},
+		ResultPaths: []*ResultPath{
+			{Name: "output1", Path: "/output1"},
+		},
+		Meta: map[string]string{"key": "value"},
+		Env:  map[string]string{"ENV_VAR": "value"},
+	}
+
+	cpy := original.Copy()
+
+	suite.Equal(original, cpy, "The task and its copy should be deeply equal")
+	suite.NotSame(original, cpy, "The task and its copy should not be the same instance")
+
+	// Ensure nested objects are deeply copied
+	suite.NotSame(original.Engine, cpy.Engine, "The Engine in the task and its copy should not be the same instance")
+	suite.NotSame(original.Publisher, cpy.Publisher, "The Publisher in the task and its copy should not be the same instance")
+	for i := range original.InputSources {
+		suite.NotSame(original.InputSources[i], cpy.InputSources[i], "The InputSources in the task and its copy should not be the same instance")
+	}
+	for i := range original.ResultPaths {
+		suite.NotSame(original.ResultPaths[i], cpy.ResultPaths[i], "The ResultPaths in the task and its copy should not be the same instance")
+	}
+
+	// Ensure it's a deep copy by modifying the copy
+	cpy.Name = "modified-task"
+	cpy.Meta["new_key"] = "new_value"
+	cpy.Env["NEW_ENV_VAR"] = "new_value"
+
+	suite.NotEqual(original.Name, cpy.Name)
+	suite.NotEqual(original.Meta, cpy.Meta)
+	suite.NotEqual(original.Env, cpy.Env)
+}
+
+func (suite *TaskTestSuite) TestAllStorageTypes() {
+	task := &Task{
+		InputSources: []*InputSource{
+			{Source: &SpecConfig{Type: "s3"}},
+			{Source: &SpecConfig{Type: "url"}},
+			{Source: &SpecConfig{Type: "s3"}}, // Duplicate to test uniqueness
+		},
+	}
+
+	storageTypes := task.AllStorageTypes()
+	suite.ElementsMatch([]string{"s3", "url"}, storageTypes)
+	suite.Len(storageTypes, 2, "Should return only unique storage types")
+}