Adding explicit job_runs dependency ordering

acceptance/bundle/refschema/out.fields.txt

@@ -914,6 +914,8 @@ resources.internal_immutable_snapshots.*.remote_root	string	INPUT
 resources.internal_immutable_snapshots.*.zip_path	string	INPUT	STATE
 resources.job_runs.*.dbt_commands	[]string	ALL
 resources.job_runs.*.dbt_commands[*]	string	ALL
+resources.job_runs.*.depends_on	[]string	INPUT	STATE
+resources.job_runs.*.depends_on[*]	string	INPUT	STATE
 resources.job_runs.*.id	string	INPUT
 resources.job_runs.*.idempotency_token	string	ALL
 resources.job_runs.*.jar_params	[]string	ALL

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

@@ -10,6 +10,18 @@ resources:
           notebook_task:
             notebook_path: /Workspace/test
 
+    downstream_job:
+      name: downstream-job
+      tasks:
+        - task_key: main
+          notebook_task:
+            notebook_path: /Workspace/test
+
   job_runs:
     my_run:
       job_id: ${resources.jobs.my_job.id}
+
+    downstream_run:
+      job_id: ${resources.jobs.downstream_job.id}
+      depends_on:
+        - ${resources.job_runs.my_run.id}

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

@@ -2,6 +2,7 @@
 === the deploy stops waiting before the run finishes
 >>> errcode [CLI] bundle deploy
 Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-interrupted-run/default/files...
+Created jobs.downstream_job
 Created jobs.my_job
 Error: cannot create resources.job_runs.my_run: waiting after creating id=[MY_RUN_ID]: Fault injected by test. (403 INJECTED)
 
@@ -10,6 +11,8 @@ HTTP Status: 403 Forbidden
 API error_code: INJECTED
 API message: Fault injected by test.
 
+Error: cannot create resources.job_runs.downstream_run: dependency failed: resources.job_runs.my_run
+
 Files: 3 uploaded, 0 deleted
 
 Exit code: 1
@@ -31,51 +34,92 @@ Exit code: 1
   }
 }
 
-=== a run that is still going is skipped
->>> jq .plan["resources.job_runs.my_run"] tmp.plan.json
+=== a run that is still going is skipped and blocks its dependent
+>>> jq {upstream: .plan["resources.job_runs.my_run"], downstream: .plan["resources.job_runs.downstream_run"]} tmp.plan.json
 {
-  "depends_on": [
-    {
-      "node": "resources.jobs.my_job",
-      "label": "${resources.jobs.my_job.id}"
-    }
-  ],
-  "action": "skip",
-  "remote_state": {
-    "job_id": [MY_JOB_ID],
-    "run_id": [MY_RUN_ID],
-    "run_name": "my-job",
-    "run_page_url": "[DATABRICKS_URL]/jobs/[MY_JOB_ID]/runs/[MY_RUN_ID]?o=[NUMID]",
-    "run_type": "JOB_RUN",
-    "state": {
-      "life_cycle_state": "RUNNING"
+  "upstream": {
+    "depends_on": [
+      {
+        "node": "resources.jobs.my_job",
+        "label": "${resources.jobs.my_job.id}"
+      }
+    ],
+    "action": "skip",
+    "remote_state": {
+      "job_id": [MY_JOB_ID],
+      "run_id": [MY_RUN_ID],
+      "run_name": "my-job",
+      "run_page_url": "[DATABRICKS_URL]/jobs/[MY_JOB_ID]/runs/[MY_RUN_ID]?o=[NUMID]",
+      "run_type": "JOB_RUN",
+      "state": {
+        "life_cycle_state": "RUNNING"
+      }
+    },
+    "changes": {
+      "result_state": {
+        "action": "skip",
+        "reason": "run in progress",
+        "new": "SUCCESS"
+      }
     }
   },
-  "changes": {
-    "result_state": {
-      "action": "skip",
-      "reason": "run in progress",
-      "new": "SUCCESS"
+  "downstream": {
+    "depends_on": [
+      {
+        "node": "resources.job_runs.my_run",
+        "label": "${resources.job_runs.my_run.id}"
+      },
+      {
+        "node": "resources.jobs.downstream_job",
+        "label": "${resources.jobs.downstream_job.id}"
+      }
+    ],
+    "action": "create",
+    "new_state": {
+      "value": {
+        "depends_on": [
+          "[MY_RUN_ID]"
+        ],
+        "job_id": [NUMID],
+        "result_state": "SUCCESS"
+      }
     }
   }
 }
 
-=== Badness: the next deploy does not wait for that run
+=== the next deploy resumes waiting before starting the dependent run
 >>> [CLI] bundle deploy --plan tmp.plan.json
 Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-interrupted-run/default/files...
+Output from job_runs.my_run: id=[MY_RUN_ID]: Run URL: [DATABRICKS_URL]/jobs/[MY_JOB_ID]/runs/[MY_RUN_ID]?o=[NUMID]
+Output from job_runs.my_run: id=[MY_RUN_ID]: SUCCESS
+Output from job_runs.downstream_run: id=[NUMID]: Run URL: [DATABRICKS_URL]/jobs/[NUMID]/runs/[NUMID]?o=[NUMID]
+Output from job_runs.downstream_run: id=[NUMID]: SUCCESS
+Created job_runs.downstream_run
 Files: 1 uploaded, 0 deleted
-Resources: 0 created, 0 changed, 0 deleted, 2 unchanged
+Resources: 1 created, 0 changed, 0 deleted, 3 unchanged
 
-=== the same run is still tracked, and nothing was cancelled, deleted or triggered
+=== both successful runs are tracked
 >>> read_state.py job_runs my_run id result_state
 job_runs my_run id='[MY_RUN_ID]' result_state=None
 
->>> print_requests.py //jobs
+>>> read_state.py job_runs downstream_run id depends_on result_state
+job_runs downstream_run id='[NUMID]' depends_on=['[MY_RUN_ID]'] result_state=None
+
+>>> print_requests.py //jobs/run-now
+{
+  "method": "POST",
+  "path": "/api/2.2/jobs/run-now",
+  "body": {
+    "idempotency_token": "[UUID]",
+    "job_id": [NUMID]
+  }
+}
 
 >>> [CLI] bundle destroy --auto-approve
 The following resources will be deleted:
+  delete resources.jobs.downstream_job
   delete resources.jobs.my_job
 
 All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/job-runs-interrupted-run/default
 
-Destroy: 1 deleted
+Destroy: 2 deleted

acceptance/bundle/resources/job_runs/interrupted_run/script

@@ -17,15 +17,16 @@ trace read_id.py my_job
 trace read_id.py my_run
 trace print_requests.py //jobs/run-now
 
-title "a run that is still going is skipped"
+title "a run that is still going is skipped and blocks its dependent"
 $CLI bundle plan -o json > tmp.plan.json
-trace jq '.plan["resources.job_runs.my_run"]' tmp.plan.json
+trace jq '{upstream: .plan["resources.job_runs.my_run"], downstream: .plan["resources.job_runs.downstream_run"]}' tmp.plan.json
 
 # Replay the plan captured while the run was still going: a fresh plan would
 # re-read it, and the testserver would settle it on that GET.
-title "Badness: the next deploy does not wait for that run"
+title "the next deploy resumes waiting before starting the dependent run"
 trace $CLI bundle deploy --plan tmp.plan.json
 
-title "the same run is still tracked, and nothing was cancelled, deleted or triggered"
+title "both successful runs are tracked"
 trace read_state.py job_runs my_run id result_state
-trace print_requests.py //jobs
+trace read_state.py job_runs downstream_run id depends_on result_state
+trace print_requests.py //jobs/run-now

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

@@ -1,4 +1,3 @@
 # The interruption is staged by injecting a fault into the testserver, so this
 # stays off cloud.
-Badness = "An in-progress run is planned as skip, so a deploy after an interrupted wait does not resume waiting for it to finish."
 Ignore = ["tmp.plan.json"]

bundle/config/resources/job_run.go

@@ -21,6 +21,9 @@ type JobRun struct {
 	ID string `json:"id,omitempty" bundle:"readonly"`
 	jobs.RunNow
 
+	// DependsOn lists job runs that must succeed before this run starts.
+	DependsOn []string `json:"depends_on,omitempty"`
+
 	// Lifecycle shadows BaseResource.Lifecycle so job_runs can set triggers.
 	Lifecycle *JobRunLifecycle `json:"lifecycle,omitempty"`
 

bundle/config/validate/validate_job_run_dependencies.go

@@ -0,0 +1,78 @@
+package validate
+
+import (
+	"context"
+	"fmt"
+	"maps"
+	"slices"
+
+	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/libs/diag"
+	"github.com/databricks/cli/libs/dyn"
+	"github.com/databricks/cli/libs/dyn/dynvar"
+)
+
+func ValidateJobRunDependencies() bundle.ReadOnlyMutator {
+	return &validateJobRunDependencies{}
+}
+
+type validateJobRunDependencies struct{ bundle.RO }
+
+func (v *validateJobRunDependencies) Name() string {
+	return "validate:job_run_dependencies"
+}
+
+func (v *validateJobRunDependencies) Apply(_ context.Context, b *bundle.Bundle) diag.Diagnostics {
+	var diags diag.Diagnostics
+
+	for _, name := range slices.Sorted(maps.Keys(b.Config.Resources.JobRuns)) {
+		jobRun := b.Config.Resources.JobRuns[name]
+		if jobRun == nil {
+			continue
+		}
+
+		for i, dependency := range jobRun.DependsOn {
+			path := dyn.NewPath(
+				dyn.Key("resources"),
+				dyn.Key("job_runs"),
+				dyn.Key(name),
+				dyn.Key("depends_on"),
+				dyn.Index(i),
+			)
+			refPath, ok := dynvar.PureReferenceToPath(dependency)
+			if !ok || !isJobRunIDReference(refPath) {
+				diags = append(diags, invalidJobRunDependency(b, path))
+				continue
+			}
+
+			target := refPath[2].Key()
+			if _, ok := b.Config.Resources.JobRuns[target]; !ok {
+				diags = append(diags, diag.Diagnostic{
+					Severity:  diag.Error,
+					Summary:   fmt.Sprintf("depends_on references undefined job run %q", target),
+					Paths:     []dyn.Path{path},
+					Locations: b.Config.GetLocations(path.String()),
+				})
+			}
+		}
+	}
+
+	return diags
+}
+
+func isJobRunIDReference(path dyn.Path) bool {
+	return len(path) == 4 &&
+		path[0].Key() == "resources" &&
+		path[1].Key() == "job_runs" &&
+		path[2].Key() != "" &&
+		path[3].Key() == "id"
+}
+
+func invalidJobRunDependency(b *bundle.Bundle, path dyn.Path) diag.Diagnostic {
+	return diag.Diagnostic{
+		Severity:  diag.Error,
+		Summary:   "depends_on entries must be job run ID references, for example ${resources.job_runs.prepare.id}",
+		Paths:     []dyn.Path{path},
+		Locations: b.Config.GetLocations(path.String()),
+	}
+}

bundle/config/validate/validate_job_run_dependencies_test.go

@@ -0,0 +1,75 @@
+package validate
+
+import (
+	"testing"
+
+	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/bundle/config"
+	"github.com/databricks/cli/bundle/config/resources"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestValidateJobRunDependencies(t *testing.T) {
+	tests := []struct {
+		name       string
+		dependency string
+		summary    string
+	}{
+		{
+			name:       "job run id reference",
+			dependency: "${resources.job_runs.prepare.id}",
+		},
+		{
+			name:       "literal name",
+			dependency: "prepare",
+			summary:    "depends_on entries must be job run ID references, for example ${resources.job_runs.prepare.id}",
+		},
+		{
+			name:       "job definition reference",
+			dependency: "${resources.jobs.prepare.id}",
+			summary:    "depends_on entries must be job run ID references, for example ${resources.job_runs.prepare.id}",
+		},
+		{
+			name:       "job run outcome reference",
+			dependency: "${resources.job_runs.prepare.state.result_state}",
+			summary:    "depends_on entries must be job run ID references, for example ${resources.job_runs.prepare.id}",
+		},
+		{
+			name:       "reference with extra text",
+			dependency: "run ${resources.job_runs.prepare.id}",
+			summary:    "depends_on entries must be job run ID references, for example ${resources.job_runs.prepare.id}",
+		},
+		{
+			name:       "undefined job run",
+			dependency: "${resources.job_runs.missing.id}",
+			summary:    `depends_on references undefined job run "missing"`,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			b := &bundle.Bundle{
+				Config: config.Root{
+					Resources: config.Resources{
+						JobRuns: map[string]*resources.JobRun{
+							"prepare": {},
+							"publish": {DependsOn: []string{tt.dependency}},
+						},
+					},
+				},
+			}
+
+			diags := ValidateJobRunDependencies().Apply(t.Context(), b)
+			if tt.summary == "" {
+				assert.Empty(t, diags)
+				return
+			}
+
+			require.Len(t, diags, 1)
+			assert.Equal(t, tt.summary, diags[0].Summary)
+			require.Len(t, diags[0].Paths, 1)
+			assert.Equal(t, "resources.job_runs.publish.depends_on[0]", diags[0].Paths[0].String())
+		})
+	}
+}