Add idempotency_token to job_runs resource

.nextchanges/bundles/job-runs-idempotency-wait.md

@@ -0,0 +1 @@
+direct: the experimental `job_runs` resource now waits for the run to finish, exposes its output fields (e.g. `${resources.job_runs.<name>.state.result_state}`), and dedupes runs on deploy retries.

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

@@ -68,6 +68,7 @@ Resources:
   "method": "POST",
   "path": "/api/2.2/jobs/run-now",
   "body": {
+    "idempotency_token": "[IDEMPOTENCY_TOKEN]",
     "job_id": [MY_JOB_ID]
   }
 }

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

@@ -0,0 +1,15 @@
+bundle:
+  name: job-runs-idempotent-recreate
+
+resources:
+  jobs:
+    my_job:
+      name: my-job
+      tasks:
+        - task_key: main
+          notebook_task:
+            notebook_path: /Workspace/test
+
+  job_runs:
+    my_run:
+      job_id: ${resources.jobs.my_job.id}

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

@@ -0,0 +1,70 @@
+
+=== initial deploy triggers the run and waits for it to finish
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-idempotent-recreate/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!
+
+=== retry mid-run after a lost run id (deploy crashed before the id was recorded)
+>>> [CLI] bundle plan -o json
+{
+  "depends_on": [
+    {
+      "node": "resources.jobs.my_job",
+      "label": "${resources.jobs.my_job.id}"
+    }
+  ],
+  "action": "create",
+  "new_state": {
+    "value": {
+      "job_id": [NUMID]
+    }
+  }
+}
+
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-idempotent-recreate/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!
+
+=== retry mid-run after the run id was recorded (deploy crashed during the wait)
+>>> [CLI] bundle plan -o json
+"skip"
+
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-idempotent-recreate/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!
+
+=== run-now was issued exactly twice, both with the same token, and no duplicate run was created
+>>> print_requests.py //jobs/run-now
+{
+  "method": "POST",
+  "path": "/api/2.2/jobs/run-now",
+  "body": {
+    "idempotency_token": "[IDEMPOTENCY_TOKEN]",
+    "job_id": [NUMID]
+  }
+}
+{
+  "method": "POST",
+  "path": "/api/2.2/jobs/run-now",
+  "body": {
+    "idempotency_token": "[IDEMPOTENCY_TOKEN]",
+    "job_id": [NUMID]
+  }
+}
+run id stable across both retries ([NUMID]): the idempotency token reused the existing run
+
+>>> [CLI] bundle destroy --auto-approve
+The following resources will be deleted:
+  delete resources.job_runs.my_run
+  delete resources.jobs.my_job
+
+All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/job-runs-idempotent-recreate/default
+
+Deleting files...
+Destroy complete!

acceptance/bundle/resources/job_runs/idempotent_recreate/script

@@ -0,0 +1,40 @@
+cleanup() {
+    trace $CLI bundle destroy --auto-approve
+    rm -f out.requests.txt
+}
+trap cleanup EXIT
+
+STATE=.databricks/bundle/default/resources.json
+run_id() {
+    jq -r '.state["resources.job_runs.my_run"].__id__' "$STATE"
+}
+
+title "initial deploy triggers the run and waits for it to finish"
+trace $CLI bundle deploy
+before=$(run_id)
+
+title "retry mid-run after a lost run id (deploy crashed before the id was recorded)"
+# Simulates a deploy that fired run-now but crashed before saving the run id.
+# job_id (and thus the idempotency token) is unchanged, so the retried run-now
+# dedupes to the existing run instead of starting a second one.
+jq 'del(.state["resources.job_runs.my_run"])' "$STATE" > "$STATE.new"
+mv "$STATE.new" "$STATE"
+trace $CLI bundle plan -o json | jq '.plan["resources.job_runs.my_run"]'
+trace $CLI bundle deploy
+after_lost=$(run_id)
+
+title "retry mid-run after the run id was recorded (deploy crashed during the wait)"
+# The run id is still in state, so the retry is a no-op: no second run-now.
+trace $CLI bundle plan -o json | jq '.plan["resources.job_runs.my_run"].action // "none"'
+trace $CLI bundle deploy
+after_kept=$(run_id)
+
+title "run-now was issued exactly twice, both with the same token, and no duplicate run was created"
+# Two run-now calls total (initial + lost-id retry); the recorded-id retry adds
+# none. All share one token, so only one run ever exists.
+trace print_requests.py //jobs/run-now
+if [ "$before" = "$after_lost" ] && [ "$after_lost" = "$after_kept" ]; then
+    echo "run id stable across both retries ($before): the idempotency token reused the existing run"
+else
+    echo "run id changed ($before -> $after_lost -> $after_kept): a duplicate run was created"
+fi

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

@@ -11,6 +11,7 @@ Deployment complete!
   "method": "POST",
   "path": "/api/2.2/jobs/run-now",
   "body": {
+    "idempotency_token": "[IDEMPOTENCY_TOKEN]",
     "job_id": [NUMID],
     "job_parameters": {
       "env": "prod"

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

@@ -30,6 +30,7 @@ Resources:
   "method": "POST",
   "path": "/api/2.2/jobs/run-now",
   "body": {
+    "idempotency_token": "[IDEMPOTENCY_TOKEN]",
     "job_id": [MY_JOB_ID],
     "job_parameters": {
       "env": "dev"
@@ -67,7 +68,8 @@ Resources:
     "run_page_url": "[DATABRICKS_URL]/?o=[NUMID]#job/[MY_JOB_ID]/run/[NUMID]",
     "run_type": "JOB_RUN",
     "state": {
-      "life_cycle_state": "RUNNING"
+      "life_cycle_state": "TERMINATED",
+      "result_state": "SUCCESS"
     }
   },
   "changes": {
@@ -118,6 +120,7 @@ Resources:
   "method": "POST",
   "path": "/api/2.2/jobs/run-now",
   "body": {
+    "idempotency_token": "[IDEMPOTENCY_TOKEN]",
     "job_id": [MY_JOB_ID],
     "job_parameters": {
       "env": "prod"

acceptance/bundle/resources/job_runs/test.toml

@@ -0,0 +1,11 @@
+# job_runs is a direct-engine-only resource; the Terraform provider has no
+# equivalent, so restrict the matrix to direct.
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
+RecordRequests = true
+
+# idempotency_token is a SHA-256 of the RunNow request, and the fake assigns
+# job_id from a time-based counter, so the hash differs every run. Mask it so the
+# run-now request goldens stay stable.
+[[Repls]]
+Old = '[0-9a-f]{64}'
+New = '[IDEMPOTENCY_TOKEN]'

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

@@ -0,0 +1,26 @@
+bundle:
+  name: job-runs-wait-output
+
+resources:
+  jobs:
+    my_job:
+      name: my-job
+      tasks:
+        - task_key: main
+          notebook_task:
+            notebook_path: /Workspace/test
+
+    # Reads the run's output. Must not be my_job (the run's target), or the
+    # reference would cycle: my_run already depends on my_job.id.
+    downstream_job:
+      name: downstream-job
+      tags:
+        run_result: ${resources.job_runs.my_run.state.result_state}
+      tasks:
+        - task_key: main
+          notebook_task:
+            notebook_path: /Workspace/test
+
+  job_runs:
+    my_run:
+      job_id: ${resources.jobs.my_job.id}

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

@@ -0,0 +1,84 @@
+
+=== deploy waits for the run to finish, then the downstream job reads its result_state
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-wait-output/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!
+
+=== the downstream job was created with the run's result_state resolved into its tag
+>>> print_requests.py //jobs/create
+{
+  "method": "POST",
+  "path": "/api/2.2/jobs/create",
+  "body": {
+    "deployment": {
+      "kind": "BUNDLE",
+      "metadata_file_path": "/Workspace/Users/[USERNAME]/.bundle/job-runs-wait-output/default/state/metadata.json"
+    },
+    "edit_mode": "UI_LOCKED",
+    "format": "MULTI_TASK",
+    "max_concurrent_runs": 1,
+    "name": "my-job",
+    "queue": {
+      "enabled": true
+    },
+    "tasks": [
+      {
+        "notebook_task": {
+          "notebook_path": "/Workspace/test"
+        },
+        "task_key": "main"
+      }
+    ]
+  }
+}
+{
+  "method": "POST",
+  "path": "/api/2.2/jobs/create",
+  "body": {
+    "deployment": {
+      "kind": "BUNDLE",
+      "metadata_file_path": "/Workspace/Users/[USERNAME]/.bundle/job-runs-wait-output/default/state/metadata.json"
+    },
+    "edit_mode": "UI_LOCKED",
+    "format": "MULTI_TASK",
+    "max_concurrent_runs": 1,
+    "name": "downstream-job",
+    "queue": {
+      "enabled": true
+    },
+    "tags": {
+      "run_result": "SUCCESS"
+    },
+    "tasks": [
+      {
+        "notebook_task": {
+          "notebook_path": "/Workspace/test"
+        },
+        "task_key": "main"
+      }
+    ]
+  }
+}
+
+=== redeploy is a no-op: the resolved result_state tag is stable, not perpetual drift
+>>> [CLI] bundle plan -o json
+"skip"
+
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-wait-output/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!
+
+>>> [CLI] bundle destroy --auto-approve
+The following resources will be deleted:
+  delete resources.job_runs.my_run
+  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-wait-output/default
+
+Deleting files...
+Destroy complete!

acceptance/bundle/resources/job_runs/wait_output/script

@@ -0,0 +1,18 @@
+cleanup() {
+    trace $CLI bundle destroy --auto-approve
+    rm -f out.requests.txt
+}
+trap cleanup EXIT
+
+title "deploy waits for the run to finish, then the downstream job reads its result_state"
+trace $CLI bundle deploy
+
+title "the downstream job was created with the run's result_state resolved into its tag"
+# A concrete SUCCESS tag proves the run finished and WaitAfterCreate published
+# its output before the downstream job was created.
+trace print_requests.py //jobs/create
+
+title "redeploy is a no-op: the resolved result_state tag is stable, not perpetual drift"
+# The resolved tag must not read back as drift, or the job recreates every deploy.
+trace $CLI bundle plan -o json | jq '.plan["resources.jobs.downstream_job"].action // "none"'
+trace $CLI bundle deploy

bundle/config/validate/validate_job_run_idempotency_token.go

@@ -0,0 +1,48 @@
+package validate
+
+import (
+	"cmp"
+	"context"
+	"slices"
+
+	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/libs/diag"
+	"github.com/databricks/cli/libs/dyn"
+)
+
+func ValidateJobRunIdempotencyToken() bundle.ReadOnlyMutator {
+	return &validateJobRunIdempotencyToken{}
+}
+
+type validateJobRunIdempotencyToken struct{ bundle.RO }
+
+func (v *validateJobRunIdempotencyToken) Name() string {
+	return "validate:validate_job_run_idempotency_token"
+}
+
+func (v *validateJobRunIdempotencyToken) Apply(_ context.Context, b *bundle.Bundle) diag.Diagnostics {
+	var diags diag.Diagnostics
+
+	// idempotency_token is computed automatically from the run configuration
+	// (SHA-256 of the RunNow request) so retries dedupe. A user-provided value
+	// would be overwritten, so reject it up front.
+	for name, jr := range b.Config.Resources.JobRuns {
+		if jr.IdempotencyToken == "" {
+			continue
+		}
+		path := "resources.job_runs." + name + ".idempotency_token"
+		diags = append(diags, diag.Diagnostic{
+			Severity:  diag.Error,
+			Summary:   "idempotency_token is computed automatically and must not be set in bundle configuration",
+			Paths:     []dyn.Path{dyn.MustPathFromString(path)},
+			Locations: b.Config.GetLocations(path),
+		})
+	}
+
+	// Map iteration order is randomized; sort by path for stable output.
+	slices.SortFunc(diags, func(x, y diag.Diagnostic) int {
+		return cmp.Compare(x.Paths[0].String(), y.Paths[0].String())
+	})
+
+	return diags
+}