testserver: roll task outcomes up into the run state

acceptance/bundle/resources/job_runs/failed_run/databricks.yml

@@ -6,9 +6,17 @@ resources:
     my_job:
       name: my-job
       tasks:
+        # The test server runs this locally; the script exits non-zero, which
+        # fails the task and with it the run.
         - task_key: main
-          notebook_task:
-            notebook_path: /Workspace/missing-notebook
+          spark_python_task:
+            python_file: ./fail.py
+          environment_key: default
+
+      environments:
+        - environment_key: default
+          spec:
+            client: "2"
 
     # Depends on my_run's result_state, so the failing run aborts the deploy
     # before this job is created.

acceptance/bundle/resources/job_runs/failed_run/fail.py

@@ -0,0 +1,4 @@
+import sys
+
+print("intentional failure", file=sys.stderr)
+sys.exit(1)

acceptance/bundle/resources/job_runs/failed_run/output.txt

@@ -3,9 +3,14 @@
 >>> [CLI] bundle deploy
 Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-failed-run/default/files...
 Deploying resources...
+resources.job_runs.my_run: Run URL: [DATABRICKS_URL]/jobs/[NUMID]/runs/[NUMID]?o=[NUMID]
+resources.job_runs.my_run: [TIMESTAMP] "my-job" RUNNING
 resources.job_runs.my_run: [TIMESTAMP] "my-job" TERMINATED FAILED task main failed
 Error: cannot create resources.job_runs.my_run: job run [NUMID] did not succeed: FAILED: task main failed
-task "main": notebook not found in workspace: /Workspace/missing-notebook
+task "main": spark python task execution failed: exit status 1
+intentional failure
+
+run page: [DATABRICKS_URL]/jobs/[NUMID]/runs/[NUMID]?o=[NUMID]
 redeploying reports this same run; set rerun_token to a new value to run the job again
 
 Error: cannot create resources.jobs.downstream_job: dependency failed: resources.job_runs.my_run

acceptance/bundle/resources/job_runs/failed_run/test.toml

@@ -1,31 +1,2 @@
 # The deploy fails mid-way, leaving local deployment state behind.
 Ignore = [".databricks"]
-
-# The fake workspace reports every run as succeeded, so a failing run has to be
-# stubbed. run_page_url is omitted because a static stub cannot know the test
-# server's address; wait_output covers the run page link.
-[[Server]]
-Pattern = "GET /api/2.2/jobs/runs/get"
-Response.Body = '''
-{
-  "run_id": 12345678,
-  "job_id": 87654321,
-  "run_name": "my-job",
-  "state": {
-    "life_cycle_state": "TERMINATED",
-    "result_state": "FAILED",
-    "state_message": "task main failed"
-  },
-  "tasks": [
-    {
-      "task_key": "main",
-      "run_id": 12345679,
-      "state": {"life_cycle_state": "TERMINATED", "result_state": "FAILED"}
-    }
-  ]
-}
-'''
-
-[[Server]]
-Pattern = "GET /api/2.2/jobs/runs/get-output"
-Response.Body = '{"error": "notebook not found in workspace: /Workspace/missing-notebook"}'

libs/testserver/jobs.go

@@ -20,6 +20,11 @@ import (
 
 const missingJobGitProviderMessage = "git_source.git_provider must be one of: github,gitlab,bitbucketcloud,gitlabenterpriseedition,bitbucketserver,azuredevopsservices,githubenterprise,awscodecommit"
 
+// errNoCodeInWorkspace marks a task there is nothing to execute for, e.g.
+// because an immutable deployment uploaded the code as a snapshot zip this
+// server never unpacks. The gap is here, not in the job, so the task succeeds.
+var errNoCodeInWorkspace = errors.New("task code is not in the workspace")
+
 // venvPython returns the path to the Python executable in a venv.
 // On Unix: venv/bin/python
 // On Windows: venv\Scripts\python.exe
@@ -401,12 +406,15 @@ func (s *FakeWorkspace) JobsRunNow(req Request) Response {
 				logs, err = s.executeSparkPythonTask(t)
 			}
 
-			if err != nil {
+			switch {
+			case errors.Is(err, errNoCodeInWorkspace):
+				// Nothing ran, so the task keeps its SUCCESS state.
+			case err != nil:
 				taskRun.State.ResultState = jobs.RunResultStateFailed
 				s.JobRunOutputs[taskRunId] = jobs.RunOutput{
 					Error: err.Error(),
 				}
-			} else if logs != "" {
+			case logs != "":
 				s.JobRunOutputs[taskRunId] = jobs.RunOutput{
 					Logs: logs,
 				}
@@ -617,7 +625,7 @@ func (s *FakeWorkspace) executePythonWheelTask(jobSettings *jobs.JobSettings, ta
 		}
 		data := s.files[whlPath].Data
 		if len(data) == 0 {
-			return "", fmt.Errorf("wheel file not found in workspace: %s", whlPath)
+			return "", fmt.Errorf("%w: wheel file not found in workspace: %s", errNoCodeInWorkspace, whlPath)
 		}
 		localPath := filepath.Join(env.dir, filepath.Base(whlPath))
 		if err := os.WriteFile(localPath, data, 0o644); err != nil {
@@ -636,7 +644,7 @@ func (s *FakeWorkspace) executePythonWheelTask(jobSettings *jobs.JobSettings, ta
 	}
 
 	if len(env.installedLibs) == 0 {
-		return "", errors.New("no wheel libraries found in task")
+		return "", fmt.Errorf("%w: no wheel libraries found in task", errNoCodeInWorkspace)
 	}
 
 	// Run the entry point using runpy with sys.argv[0] set to the package name,
@@ -682,7 +690,7 @@ func (s *FakeWorkspace) executeNotebookTask(task jobs.Task, notebookParams map[s
 		notebookData = s.files[notebookPath+".py"].Data
 	}
 	if len(notebookData) == 0 {
-		return "", fmt.Errorf("notebook not found in workspace: %s (also tried .py)", notebookPath)
+		return "", fmt.Errorf("%w: notebook not found in workspace: %s (also tried .py)", errNoCodeInWorkspace, notebookPath)
 	}
 
 	// Create a temporary Python environment for notebook execution
@@ -768,7 +776,7 @@ func (s *FakeWorkspace) executeSparkPythonTask(task jobs.Task) (string, error) {
 
 	pythonData := s.files[pythonPath].Data
 	if len(pythonData) == 0 {
-		return "", fmt.Errorf("python file not found in workspace: %s", pythonPath)
+		return "", fmt.Errorf("%w: python file not found in workspace: %s", errNoCodeInWorkspace, pythonPath)
 	}
 
 	env, cleanup, err := s.getOrCreateClusterEnv(task)
@@ -866,6 +874,30 @@ func sparkVersionToPython(task jobs.Task) string {
 	return "3.10"
 }
 
+// terminateRun completes the run, rolling task outcomes up into the run-level
+// state the way the Jobs API does: one failed task fails the whole run.
+func terminateRun(run *jobs.Run) {
+	for i := range run.Tasks {
+		// Tasks that were never executed (jobs/runs/submit) are still running.
+		if run.Tasks[i].State.LifeCycleState != jobs.RunLifeCycleStateTerminated {
+			run.Tasks[i].State.LifeCycleState = jobs.RunLifeCycleStateTerminated
+			run.Tasks[i].State.ResultState = jobs.RunResultStateSuccess
+		}
+	}
+
+	run.State = &jobs.RunState{
+		LifeCycleState: jobs.RunLifeCycleStateTerminated,
+		ResultState:    jobs.RunResultStateSuccess,
+	}
+	for _, task := range run.Tasks {
+		if task.State.ResultState != jobs.RunResultStateSuccess {
+			run.State.ResultState = task.State.ResultState
+			run.State.StateMessage = fmt.Sprintf("task %s failed", task.TaskKey)
+			return
+		}
+	}
+}
+
 func (s *FakeWorkspace) JobsGetRun(req Request) Response {
 	runId := req.URL.Query().Get("run_id")
 	runIdInt, err := strconv.ParseInt(runId, 10, 64)
@@ -883,19 +915,10 @@ func (s *FakeWorkspace) JobsGetRun(req Request) Response {
 		return Response{StatusCode: 404}
 	}
 
-	// Simulate cloud behavior: first poll returns RUNNING, next returns TERMINATED SUCCESS.
+	// Simulate cloud behavior: first poll returns RUNNING, next the terminal state.
 	if run.State.LifeCycleState == jobs.RunLifeCycleStateRunning {
 		// Transition stored state to TERMINATED for the next poll.
-		run.State = &jobs.RunState{
-			LifeCycleState: jobs.RunLifeCycleStateTerminated,
-			ResultState:    jobs.RunResultStateSuccess,
-		}
-		for i := range run.Tasks {
-			run.Tasks[i].State = &jobs.RunState{
-				LifeCycleState: jobs.RunLifeCycleStateTerminated,
-				ResultState:    jobs.RunResultStateSuccess,
-			}
-		}
+		terminateRun(&run)
 		s.JobRuns[runIdInt] = run
 
 		// Return RUNNING for this poll (before the transition).

libs/testserver/jobs_test.go

@@ -87,9 +87,9 @@ func TestJobsSubmit_RunReachesTerminalStateOnPoll(t *testing.T) {
 	assert.Equal(t, jobs.RunResultStateSuccess, second.State.ResultState)
 }
 
-func createJob(t *testing.T, workspace *FakeWorkspace) int64 {
+func createJob(t *testing.T, workspace *FakeWorkspace, tasks ...jobs.Task) int64 {
 	t.Helper()
-	body, err := json.Marshal(jobs.CreateJob{Name: "my-job"})
+	body, err := json.Marshal(jobs.CreateJob{Name: "my-job", Tasks: tasks})
 	require.NoError(t, err)
 
 	response := workspace.JobsCreate(Request{Body: body})
@@ -134,6 +134,60 @@ func TestJobsRunNow_IdempotencyTokenTombstonedAfterDelete(t *testing.T) {
 	assert.Contains(t, third.Body.(string), "has been deleted")
 }
 
+func terminatedTask(taskKey string, result jobs.RunResultState) jobs.RunTask {
+	return jobs.RunTask{
+		TaskKey: taskKey,
+		State: &jobs.RunState{
+			LifeCycleState: jobs.RunLifeCycleStateTerminated,
+			ResultState:    result,
+		},
+	}
+}
+
+func TestTerminateRun_FailedTaskFailsTheRun(t *testing.T) {
+	run := jobs.Run{Tasks: []jobs.RunTask{
+		terminatedTask("first", jobs.RunResultStateSuccess),
+		terminatedTask("second", jobs.RunResultStateFailed),
+	}}
+
+	terminateRun(&run)
+
+	assert.Equal(t, jobs.RunLifeCycleStateTerminated, run.State.LifeCycleState)
+	assert.Equal(t, jobs.RunResultStateFailed, run.State.ResultState)
+	assert.Equal(t, "task second failed", run.State.StateMessage)
+}
+
+func TestTerminateRun_CompletesTasksThatAreStillRunning(t *testing.T) {
+	// jobs/runs/submit records its tasks as running: they are never executed.
+	run := jobs.Run{Tasks: []jobs.RunTask{
+		{TaskKey: "main", State: &jobs.RunState{LifeCycleState: jobs.RunLifeCycleStateRunning}},
+	}}
+
+	terminateRun(&run)
+
+	assert.Equal(t, jobs.RunResultStateSuccess, run.State.ResultState)
+	assert.Empty(t, run.State.StateMessage)
+	assert.Equal(t, jobs.RunResultStateSuccess, run.Tasks[0].State.ResultState)
+}
+
+// See errNoCodeInWorkspace: a missing notebook is this server's gap, not a
+// failure of the job.
+func TestJobsGetRun_TaskWithoutCodeDoesNotFailTheRun(t *testing.T) {
+	workspace := NewFakeWorkspace("http://test", "dbapi123")
+	jobID := createJob(t, workspace, jobs.Task{
+		TaskKey:      "main",
+		NotebookTask: &jobs.NotebookTask{NotebookPath: "/missing-notebook"},
+	})
+
+	response := runNow(t, workspace, jobs.RunNow{JobId: jobID})
+	require.Equal(t, 0, response.StatusCode)
+	runID := response.Body.(jobs.RunNowResponse).RunId
+
+	// The first poll reports RUNNING, the second the terminal state.
+	require.Equal(t, jobs.RunLifeCycleStateRunning, getRun(t, workspace, runID).State.LifeCycleState)
+	assert.Equal(t, jobs.RunResultStateSuccess, getRun(t, workspace, runID).State.ResultState)
+}
+
 func TestJobsSubmit_RejectsInvalidGitProvider(t *testing.T) {
 	workspace := NewFakeWorkspace("http://test", "dbapi123")