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

Issue:
Default publisher is not getting applied
Submitting a job with outputs and without a publisher defined is expected to have the default publisher applied in the requester side, if any, or is failed. 

However, this is not the case as we are doing client side validation and fail the job early before submission if it has result paths defined and no publisher

This has been failing since v1.4.0

```
# run node with local default publisher
bacalhau-1.4.0 serve --node-type requester,compute --repo ~/bacalhau-versions/v1.4.0/data-dir --default-publisher local

# run job with results but no publisher
→ bacalhau docker run --output outputs:/outputs ubuntu -- sh -c "echo hello > /outputs/hello.txt"
building job spec: failed to create job: publisher must be set if result paths are set

# verify job is passing if a publisher is defined
bacalhau docker run --output outputs:/outputs -p local ubuntu -- sh -c "echo hello > /outputs/hello.txt"
```

The same applies for declarative submissions, such as 
```
# publisher-default.yaml
Name: Default publisher
Type: batch
Tasks:
  - Name: main
    Engine:
      Type: docker
      Params:
        Image: ubuntu
        Parameters:
          - sh
          - -c
          - echo hello > /outputs/hello.txt
    ResultPaths:
      - Name: outputs
        Path: /outputs
```

```
→ bacalhau job run publisher-default.yaml
the job provided is invalid: task main validation failed: publisher must be set if result paths are set
```



Discussion / hints:
The validation of tasks enforcing this behavior was introduced in https://github.com/bacalhau-project/bacalhau/pull/2802 and has thus been present since roughly [v1.1.0](https://github.com/bacalhau-project/bacalhau/releases/tag/v1.1.0).

@wdbaruni is the acceptance criteria here to 1) modify the Tasks `ValidateSubmission` method such that it is only called after the job has been transformed server side 2) modify the transformer to check if results path is defined before applying a default publisher? I believe this change may need to be landed with https://github.com/bacalhau-project/bacalhau/issues/4392 since automatically including a results path in docker jobs will cause validation of the task to fail - an is likely the reason the default was removed.
We discussed this over our 1:1. Just moving the validation from ValidateSubmission to Validate will be enough

Fix patch:
diff --git a/pkg/models/task.go b/pkg/models/task.go
index 4a5706e991..d1fc9d6691 100644
--- a/pkg/models/task.go
+++ b/pkg/models/task.go
@@ -106,12 +106,13 @@ func (t *Task) Validate() error {
 	var mErr error
 	mErr = errors.Join(mErr, t.ValidateSubmission())
 
+	if len(t.ResultPaths) > 0 && t.Publisher.IsEmpty() {
+		mErr = errors.Join(mErr, errors.New("publisher must be set if result paths are set"))
+	}
+
 	if err := t.Timeouts.Validate(); err != nil {
 		mErr = errors.Join(mErr, fmt.Errorf("task timeouts validation failed: %v", err))
 	}
-	if err := t.ResourcesConfig.Validate(); err != nil {
-		mErr = errors.Join(mErr, fmt.Errorf("task resources validation failed: %v", err))
-	}
 	return mErr
 }
 
@@ -129,15 +130,18 @@ func (t *Task) ValidateSubmission() error {
 	if err := t.Publisher.ValidateAllowBlank(); err != nil {
 		mErr = errors.Join(mErr, fmt.Errorf("publisher validation failed: %v", err))
 	}
+	if err := t.Timeouts.ValidateSubmission(); err != nil {
+		mErr = errors.Join(mErr, fmt.Errorf("task timeouts validation failed: %v", err))
+	}
+	if err := t.ResourcesConfig.Validate(); err != nil {
+		mErr = errors.Join(mErr, fmt.Errorf("task resources validation failed: %v", err))
+	}
 	if err := ValidateSlice(t.InputSources); err != nil {
 		mErr = errors.Join(mErr, fmt.Errorf("artifact validation failed: %v", err))
 	}
 	if err := ValidateSlice(t.ResultPaths); err != nil {
 		mErr = errors.Join(mErr, fmt.Errorf("output validation failed: %v", err))
 	}
-	if len(t.ResultPaths) > 0 && t.Publisher.IsEmpty() {
-		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)
diff --git a/pkg/models/timeout.go b/pkg/models/timeout.go
index d070a5ec54..67e9c848b6 100644
--- a/pkg/models/timeout.go
+++ b/pkg/models/timeout.go
@@ -54,7 +54,25 @@ func (t *TimeoutConfig) Copy() *TimeoutConfig {
 	}
 }
 
+// Validate is used to check a timeout config for reasonable configuration.
+// This is called after server side defaults are applied.
 func (t *TimeoutConfig) Validate() error {
+	if t == nil {
+		return errors.New("missing timeout config")
+	}
+	mErr := t.ValidateSubmission()
+	if t.TotalTimeout > 0 {
+		if (t.ExecutionTimeout + t.QueueTimeout) > t.TotalTimeout {
+			mErr = errors.Join(mErr, fmt.Errorf(
+				"execution timeout %s and queue timeout %s should be less than total timeout %s",
+				t.GetExecutionTimeout(), t.GetQueueTimeout(), t.GetTotalTimeout()))
+		}
+	}
+	return mErr
+}
+
+// ValidateSubmission is used to check a timeout config for reasonable configuration when it is submitted.
+func (t *TimeoutConfig) ValidateSubmission() error {
 	if t == nil {
 		return errors.New("missing timeout config")
 	}
@@ -68,12 +86,5 @@ func (t *TimeoutConfig) Validate() error {
 	if t.TotalTimeout < 0 {
 		mErr = errors.Join(mErr, fmt.Errorf("invalid total timeout value: %s", t.GetTotalTimeout()))
 	}
-	if t.TotalTimeout > 0 {
-		if (t.ExecutionTimeout + t.QueueTimeout) > t.TotalTimeout {
-			mErr = errors.Join(mErr, fmt.Errorf(
-				"execution timeout %s and queue timeout %s should be less than total timeout %s",
-				t.GetExecutionTimeout(), t.GetQueueTimeout(), t.GetTotalTimeout()))
-		}
-	}
 	return mErr
 }


Test patch:
diff --git a/pkg/models/task_test.go b/pkg/models/task_test.go
index bbe7716a5e..5ef284ad18 100644
--- a/pkg/models/task_test.go
+++ b/pkg/models/task_test.go
@@ -46,11 +46,19 @@ func (suite *TaskTestSuite) TestTaskNormalization() {
 }
 
 func (suite *TaskTestSuite) TestTaskValidation() {
+	type validationMode int
+
+	const (
+		noError validationMode = iota
+		submissionError
+		postSubmissionError
+	)
+
 	tests := []struct {
-		name    string
-		task    *Task
-		wantErr bool
-		errMsg  string
+		name           string
+		task           *Task
+		validationMode validationMode
+		errMsg         string
 	}{
 		{
 			name: "Valid task",
@@ -67,7 +75,7 @@ func (suite *TaskTestSuite) TestTaskValidation() {
 				},
 				Publisher: &SpecConfig{Type: "s3"},
 			},
-			wantErr: false,
+			validationMode: noError,
 		},
 		{
 			name: "Empty task name",
@@ -75,8 +83,8 @@ func (suite *TaskTestSuite) TestTaskValidation() {
 				Name:   "",
 				Engine: &SpecConfig{Type: "docker"},
 			},
-			wantErr: true,
-			errMsg:  "missing task name",
+			validationMode: submissionError,
+			errMsg:         "missing task name",
 		},
 		{
 			name: "Duplicate input source alias",
@@ -88,8 +96,8 @@ func (suite *TaskTestSuite) TestTaskValidation() {
 					{Alias: "input1", Target: "/input2", Source: &SpecConfig{Type: "http"}},
 				},
 			},
-			wantErr: true,
-			errMsg:  "input source with alias 'input1' already exists",
+			validationMode: submissionError,
+			errMsg:         "input source with alias 'input1' already exists",
 		},
 		{
 			name: "Duplicate input source target",
@@ -101,8 +109,8 @@ func (suite *TaskTestSuite) TestTaskValidation() {
 					{Alias: "input2", Target: "/input", Source: &SpecConfig{Type: "http"}},
 				},
 			},
-			wantErr: true,
-			errMsg:  "input source with target '/input' already exists",
+			validationMode: submissionError,
+			errMsg:         "input source with target '/input' already exists",
 		},
 		{
 			name: "Duplicate result path name",
@@ -115,8 +123,8 @@ func (suite *TaskTestSuite) TestTaskValidation() {
 				},
 				Publisher: &SpecConfig{Type: "s3"},
 			},
-			wantErr: true,
-			errMsg:  "result path with name 'output' already exists",
+			validationMode: submissionError,
+			errMsg:         "result path with name 'output' already exists",
 		},
 		{
 			name: "Duplicate result path",
@@ -129,8 +137,8 @@ func (suite *TaskTestSuite) TestTaskValidation() {
 				},
 				Publisher: &SpecConfig{Type: "s3"},
 			},
-			wantErr: true,
-			errMsg:  "result path '/output' already exists",
+			validationMode: submissionError,
+			errMsg:         "result path '/output' already exists",
 		},
 		{
 			name: "Result paths without publisher",
@@ -141,15 +149,65 @@ func (suite *TaskTestSuite) TestTaskValidation() {
 					{Name: "output", Path: "/output"},
 				},
 			},
-			wantErr: true,
-			errMsg:  "publisher must be set if result paths are set",
+			validationMode: postSubmissionError,
+			errMsg:         "publisher must be set if result paths are set",
+		},
+		{
+			name: "Misconfigured timeouts",
+			task: &Task{
+				Name:   "invalid-timeouts",
+				Engine: &SpecConfig{Type: "docker"},
+				Timeouts: &TimeoutConfig{
+					ExecutionTimeout: 100,
+					TotalTimeout:     10,
+				},
+			},
+			validationMode: postSubmissionError,
+			errMsg:         "should be less than total timeout",
+		},
+		{
+			name: "Invalid timeouts",
+			task: &Task{
+				Name:   "invalid-timeouts",
+				Engine: &SpecConfig{Type: "docker"},
+				Timeouts: &TimeoutConfig{
+					ExecutionTimeout: -1,
+				},
+			},
+			validationMode: submissionError,
+			errMsg:         "task timeouts validation failed",
+		},
+		{
+			name: "Invalid resources",
+			task: &Task{
+				Name:   "invalid-resources",
+				Engine: &SpecConfig{Type: "docker"},
+				ResourcesConfig: &ResourcesConfig{
+					CPU: "-1",
+				},
+			},
+			validationMode: submissionError,
+			errMsg:         "task resources validation failed",
 		},
 	}
 
 	for _, tt := range tests {
 		suite.Run(tt.name, func() {
+			tt.task.Normalize()
+
+			// Test ValidateSubmission()
 			err := tt.task.ValidateSubmission()
-			if tt.wantErr {
+			if tt.validationMode == submissionError {
+				suite.Error(err)
+				suite.Contains(err.Error(), tt.errMsg)
+			} else {
+				suite.NoError(err)
+			}
+
+			// Test Validate()
+			// Should always fail if ValidateSubmission() failed
+			err = tt.task.Validate()
+			if tt.validationMode != noError {
 				suite.Error(err)
 				suite.Contains(err.Error(), tt.errMsg)
 			} else {