diff --git a/.nextchanges/bundles/job-runs-idempotency-wait.md b/.nextchanges/bundles/job-runs-idempotency-wait.md new file mode 100644 index 00000000000..523b238cd59 --- /dev/null +++ b/.nextchanges/bundles/job-runs-idempotency-wait.md @@ -0,0 +1 @@ +direct: the experimental `job_runs` resource now waits for the run to finish (printing the run page URL while it waits), exposes its output fields (e.g. `${resources.job_runs..state.result_state}`), and dedupes runs on deploy retries. Set the new `rerun` field to force a fresh run of an otherwise-unchanged config. diff --git a/acceptance/bundle/refschema/out.fields.txt b/acceptance/bundle/refschema/out.fields.txt index d35bfa350f0..1705dbb32f7 100644 --- a/acceptance/bundle/refschema/out.fields.txt +++ b/acceptance/bundle/refschema/out.fields.txt @@ -861,6 +861,7 @@ resources.job_runs.*.python_params []string ALL resources.job_runs.*.python_params[*] string ALL resources.job_runs.*.queue *jobs.QueueSettings ALL resources.job_runs.*.queue.enabled bool ALL +resources.job_runs.*.rerun string ALL resources.job_runs.*.resolved_job_id int64 INPUT resources.job_runs.*.run_id int64 REMOTE resources.job_runs.*.run_name string REMOTE diff --git a/acceptance/bundle/resources/job_runs/basic/output.txt b/acceptance/bundle/resources/job_runs/basic/output.txt index 14018b93ec6..5711596fd2b 100644 --- a/acceptance/bundle/resources/job_runs/basic/output.txt +++ b/acceptance/bundle/resources/job_runs/basic/output.txt @@ -34,6 +34,7 @@ Resources: >>> [CLI] bundle deploy Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-basic/default/files... Deploying resources... +Waiting for job run to complete: [DATABRICKS_URL]/?o=[NUMID]#job/[MY_JOB_ID]/run/[MY_RUN_ID] Updating deployment state... Deployment complete! @@ -68,6 +69,7 @@ Resources: "method": "POST", "path": "/api/2.2/jobs/run-now", "body": { + "idempotency_token": "[IDEMPOTENCY_TOKEN]", "job_id": [MY_JOB_ID] } } @@ -88,12 +90,3 @@ All files and directories at the following location will be deleted: /Workspace/ Deleting files... Destroy complete! - ->>> print_requests.py //jobs/runs/delete -{ - "method": "POST", - "path": "/api/2.2/jobs/runs/delete", - "body": { - "run_id": [MY_RUN_ID] - } -} diff --git a/acceptance/bundle/resources/job_runs/basic/script b/acceptance/bundle/resources/job_runs/basic/script index bb52888ea2b..18dff26b84c 100644 --- a/acceptance/bundle/resources/job_runs/basic/script +++ b/acceptance/bundle/resources/job_runs/basic/script @@ -1,7 +1,8 @@ cleanup() { + # destroy leaves the run in place (immutable historical record), so it makes + # no jobs/runs/delete call. trace $CLI bundle destroy --auto-approve - # destroy deletes the triggered run via jobs/runs/delete (also unlinks out.requests.txt) - trace print_requests.py //jobs/runs/delete + rm -f out.requests.txt } trap cleanup EXIT diff --git a/acceptance/bundle/resources/job_runs/basic/test.toml b/acceptance/bundle/resources/job_runs/basic/test.toml deleted file mode 100644 index 4b94d8b58e9..00000000000 --- a/acceptance/bundle/resources/job_runs/basic/test.toml +++ /dev/null @@ -1,4 +0,0 @@ -# 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 diff --git a/acceptance/bundle/resources/job_runs/destroy_redeploy/databricks.yml b/acceptance/bundle/resources/job_runs/destroy_redeploy/databricks.yml new file mode 100644 index 00000000000..322aea538f5 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/destroy_redeploy/databricks.yml @@ -0,0 +1,15 @@ +bundle: + name: job-runs-destroy-redeploy + +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} diff --git a/acceptance/bundle/resources/job_runs/destroy_redeploy/out.test.toml b/acceptance/bundle/resources/job_runs/destroy_redeploy/out.test.toml new file mode 100644 index 00000000000..e90b6d5d1ba --- /dev/null +++ b/acceptance/bundle/resources/job_runs/destroy_redeploy/out.test.toml @@ -0,0 +1,3 @@ +Local = true +Cloud = false +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] diff --git a/acceptance/bundle/resources/job_runs/destroy_redeploy/output.txt b/acceptance/bundle/resources/job_runs/destroy_redeploy/output.txt new file mode 100644 index 00000000000..aaf301288e5 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/destroy_redeploy/output.txt @@ -0,0 +1,57 @@ + +=== initial deploy triggers the run +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-destroy-redeploy/default/files... +Deploying resources... +Waiting for job run to complete: [DATABRICKS_URL]/?o=[NUMID]#job/[NUMID]/run/[NUMID] +Updating deployment state... +Deployment complete! + +=== destroy removes the resource from state but leaves the run in place (no jobs/runs/delete) +>>> [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-destroy-redeploy/default + +Deleting files... +Destroy complete! + +>>> print_requests.py --keep //jobs/runs/delete + +=== redeploy after destroy succeeds and triggers a fresh run +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-destroy-redeploy/default/files... +Deploying resources... +Waiting for job run to complete: [DATABRICKS_URL]/?o=[NUMID]#job/[NUMID]/run/[NUMID] +Updating deployment state... +Deployment complete! + +>>> 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] + } +} + +>>> [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-destroy-redeploy/default + +Deleting files... +Destroy complete! diff --git a/acceptance/bundle/resources/job_runs/destroy_redeploy/script b/acceptance/bundle/resources/job_runs/destroy_redeploy/script new file mode 100644 index 00000000000..05a9f1e1161 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/destroy_redeploy/script @@ -0,0 +1,18 @@ +cleanup() { + trace $CLI bundle destroy --auto-approve + rm -f out.requests.txt +} +trap cleanup EXIT + +title "initial deploy triggers the run" +trace $CLI bundle deploy + +title "destroy removes the resource from state but leaves the run in place (no jobs/runs/delete)" +trace $CLI bundle destroy --auto-approve +trace print_requests.py --keep //jobs/runs/delete + +title "redeploy after destroy succeeds and triggers a fresh run" +# Destroy never deleted the run, so its token isn't tombstoned. The job is +# recreated with a new id, so the run config differs and a fresh run is triggered. +trace $CLI bundle deploy +trace print_requests.py //jobs/run-now diff --git a/acceptance/bundle/resources/job_runs/distinct_runs/databricks.yml b/acceptance/bundle/resources/job_runs/distinct_runs/databricks.yml new file mode 100644 index 00000000000..fc9ce302fa4 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/distinct_runs/databricks.yml @@ -0,0 +1,19 @@ +bundle: + name: job-runs-distinct-runs + +resources: + jobs: + my_job: + name: my-job + tasks: + - task_key: main + notebook_task: + notebook_path: /Workspace/test + + # Two runs with byte-identical config. Each is a distinct resource, so each + # must get its own run rather than deduping onto a shared idempotency token. + job_runs: + my_run_a: + job_id: ${resources.jobs.my_job.id} + my_run_b: + job_id: ${resources.jobs.my_job.id} diff --git a/acceptance/bundle/resources/job_runs/distinct_runs/out.test.toml b/acceptance/bundle/resources/job_runs/distinct_runs/out.test.toml new file mode 100644 index 00000000000..e90b6d5d1ba --- /dev/null +++ b/acceptance/bundle/resources/job_runs/distinct_runs/out.test.toml @@ -0,0 +1,3 @@ +Local = true +Cloud = false +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] diff --git a/acceptance/bundle/resources/job_runs/distinct_runs/output.txt b/acceptance/bundle/resources/job_runs/distinct_runs/output.txt new file mode 100644 index 00000000000..ea3447a3356 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/distinct_runs/output.txt @@ -0,0 +1,40 @@ + +=== deploy triggers a separate run for each resource +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-distinct-runs/default/files... +Deploying resources... +Waiting for job run to complete: [DATABRICKS_URL]/?o=[NUMID]#job/[NUMID]/run/[NUMID] +Waiting for job run to complete: [DATABRICKS_URL]/?o=[NUMID]#job/[NUMID]/run/[NUMID] +Updating deployment state... +Deployment complete! + +=== two run-now requests were made; each resource stored its own run id +>>> 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 ids differ: identical config did not collapse onto one run + +>>> [CLI] bundle destroy --auto-approve +The following resources will be deleted: + delete resources.job_runs.my_run_a + delete resources.job_runs.my_run_b + delete resources.jobs.my_job + +All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/job-runs-distinct-runs/default + +Deleting files... +Destroy complete! diff --git a/acceptance/bundle/resources/job_runs/distinct_runs/script b/acceptance/bundle/resources/job_runs/distinct_runs/script new file mode 100644 index 00000000000..d8f774b1ffa --- /dev/null +++ b/acceptance/bundle/resources/job_runs/distinct_runs/script @@ -0,0 +1,26 @@ +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.$1\"].__id__" "$STATE" +} + +title "deploy triggers a separate run for each resource" +trace $CLI bundle deploy +a=$(run_id my_run_a) +b=$(run_id my_run_b) + +# Both run-now bodies are identical (same job_id, masked token), so recording +# order doesn't matter. Compare ids via state, not stdout, so the golden stays +# order-independent. +title "two run-now requests were made; each resource stored its own run id" +trace print_requests.py //jobs/run-now +if [ "$a" != "$b" ]; then + echo "run ids differ: identical config did not collapse onto one run" +else + echo "run ids equal: the two resources deduped onto a single run" +fi diff --git a/acceptance/bundle/resources/job_runs/failed_run/databricks.yml b/acceptance/bundle/resources/job_runs/failed_run/databricks.yml new file mode 100644 index 00000000000..91dde0f9ab7 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/failed_run/databricks.yml @@ -0,0 +1,26 @@ +bundle: + name: job-runs-failed-run + +resources: + jobs: + my_job: + name: my-job + tasks: + - task_key: main + notebook_task: + notebook_path: /Workspace/test + + # Depends on my_run's result_state. The run finishes FAILED (see test.toml + # stub), so the deploy aborts at the run and this job is never created. + 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} diff --git a/acceptance/bundle/resources/job_runs/failed_run/out.test.toml b/acceptance/bundle/resources/job_runs/failed_run/out.test.toml new file mode 100644 index 00000000000..e90b6d5d1ba --- /dev/null +++ b/acceptance/bundle/resources/job_runs/failed_run/out.test.toml @@ -0,0 +1,3 @@ +Local = true +Cloud = false +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] diff --git a/acceptance/bundle/resources/job_runs/failed_run/output.txt b/acceptance/bundle/resources/job_runs/failed_run/output.txt new file mode 100644 index 00000000000..dd84b9e187f --- /dev/null +++ b/acceptance/bundle/resources/job_runs/failed_run/output.txt @@ -0,0 +1,21 @@ + +=== a run that finishes FAILED fails the deploy +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-failed-run/default/files... +Deploying resources... +Error: cannot create resources.job_runs.my_run: job run [NUMID] did not succeed: FAILED + +Error: cannot create resources.jobs.downstream_job: dependency failed: resources.job_runs.my_run + +Updating deployment state... + +>>> [CLI] bundle destroy --auto-approve +The following resources will be deleted: + delete resources.jobs.my_job + +All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/job-runs-failed-run/default + +Deleting files... +Destroy complete! + +Exit code: 1 diff --git a/acceptance/bundle/resources/job_runs/failed_run/script b/acceptance/bundle/resources/job_runs/failed_run/script new file mode 100644 index 00000000000..9812b06ff20 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/failed_run/script @@ -0,0 +1,10 @@ +cleanup() { + trace $CLI bundle destroy --auto-approve + rm -f out.requests.txt +} +trap cleanup EXIT + +# The run finishes FAILED, so the deploy aborts here; downstream_job depends on +# my_run and is reported as "dependency failed" rather than being created. +title "a run that finishes FAILED fails the deploy" +trace $CLI bundle deploy diff --git a/acceptance/bundle/resources/job_runs/failed_run/test.toml b/acceptance/bundle/resources/job_runs/failed_run/test.toml new file mode 100644 index 00000000000..8d54d089386 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/failed_run/test.toml @@ -0,0 +1,16 @@ +# Force my_run's GetRun to report a terminal FAILED result so we can assert a +# failed run fails the deploy. The first handler registered for a pattern wins, +# so this overrides the default runs/get handler. + +# The deploy fails mid-way, leaving local deployment state behind. +Ignore = [".databricks"] + +[[Server]] +Pattern = "GET /api/2.2/jobs/runs/get" +Response.Body = '''{ + "run_id": 1, + "state": { + "life_cycle_state": "TERMINATED", + "result_state": "FAILED" + } +}''' diff --git a/acceptance/bundle/resources/job_runs/flip_back/databricks.yml b/acceptance/bundle/resources/job_runs/flip_back/databricks.yml new file mode 100644 index 00000000000..ead9c1df65e --- /dev/null +++ b/acceptance/bundle/resources/job_runs/flip_back/databricks.yml @@ -0,0 +1,17 @@ +bundle: + name: job-runs-flip-back + +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} + job_parameters: + env: dev diff --git a/acceptance/bundle/resources/job_runs/flip_back/out.test.toml b/acceptance/bundle/resources/job_runs/flip_back/out.test.toml new file mode 100644 index 00000000000..e90b6d5d1ba --- /dev/null +++ b/acceptance/bundle/resources/job_runs/flip_back/out.test.toml @@ -0,0 +1,3 @@ +Local = true +Cloud = false +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] diff --git a/acceptance/bundle/resources/job_runs/flip_back/output.txt b/acceptance/bundle/resources/job_runs/flip_back/output.txt new file mode 100644 index 00000000000..13ca8157aa1 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/flip_back/output.txt @@ -0,0 +1,42 @@ + +=== deploy with env=dev triggers the first run +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-flip-back/default/files... +Deploying resources... +Waiting for job run to complete: [DATABRICKS_URL]/?o=[NUMID]#job/[NUMID]/run/[NUMID] +Updating deployment state... +Deployment complete! + +=== flip env dev -> prod: the config changed, so a fresh, different run is triggered +>>> update_file.py databricks.yml env: dev env: prod + +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-flip-back/default/files... +Deploying resources... +Waiting for job run to complete: [DATABRICKS_URL]/?o=[NUMID]#job/[NUMID]/run/[NUMID] +Updating deployment state... +Deployment complete! + +=== flip env prod -> dev: dedupes back to the original dev run, does not start a new one +>>> update_file.py databricks.yml env: prod env: dev + +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-flip-back/default/files... +Deploying resources... +Waiting for job run to complete: [DATABRICKS_URL]/?o=[NUMID]#job/[NUMID]/run/[NUMID] +Updating deployment state... +Deployment complete! + +=== no run was ever deleted (destroy/recreate leave runs in place), and the dev run id was reused +>>> print_requests.py --keep //jobs/runs/delete +flip-back reused the original dev run ([NUMID]); prod had its own run ([NUMID]) + +>>> [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-flip-back/default + +Deleting files... +Destroy complete! diff --git a/acceptance/bundle/resources/job_runs/flip_back/script b/acceptance/bundle/resources/job_runs/flip_back/script new file mode 100644 index 00000000000..243ac8fb7e0 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/flip_back/script @@ -0,0 +1,35 @@ +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 "deploy with env=dev triggers the first run" +trace $CLI bundle deploy +dev1=$(run_id) + +title "flip env dev -> prod: the config changed, so a fresh, different run is triggered" +trace update_file.py databricks.yml "env: dev" "env: prod" +trace $CLI bundle deploy +prod=$(run_id) + +# The dev->prod recreate left the original dev run in place (noop delete), so its +# token is still live. Flipping back to dev re-issues run-now with that token and +# dedupes onto the original dev run instead of starting a new one. +title "flip env prod -> dev: dedupes back to the original dev run, does not start a new one" +trace update_file.py databricks.yml "env: prod" "env: dev" +trace $CLI bundle deploy +dev2=$(run_id) + +title "no run was ever deleted (destroy/recreate leave runs in place), and the dev run id was reused" +trace print_requests.py --keep //jobs/runs/delete +if [ "$dev1" = "$dev2" ] && [ "$dev1" != "$prod" ]; then + echo "flip-back reused the original dev run ($dev1); prod had its own run ($prod)" +else + echo "unexpected run ids: dev1=$dev1 prod=$prod dev2=$dev2" +fi diff --git a/acceptance/bundle/resources/job_runs/idempotent_recreate/databricks.yml b/acceptance/bundle/resources/job_runs/idempotent_recreate/databricks.yml new file mode 100644 index 00000000000..6ff4be6a2ae --- /dev/null +++ b/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} diff --git a/acceptance/bundle/resources/job_runs/idempotent_recreate/out.test.toml b/acceptance/bundle/resources/job_runs/idempotent_recreate/out.test.toml new file mode 100644 index 00000000000..e90b6d5d1ba --- /dev/null +++ b/acceptance/bundle/resources/job_runs/idempotent_recreate/out.test.toml @@ -0,0 +1,3 @@ +Local = true +Cloud = false +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] diff --git a/acceptance/bundle/resources/job_runs/idempotent_recreate/output.txt b/acceptance/bundle/resources/job_runs/idempotent_recreate/output.txt new file mode 100644 index 00000000000..96e84c0b951 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/idempotent_recreate/output.txt @@ -0,0 +1,72 @@ + +=== 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... +Waiting for job run to complete: [DATABRICKS_URL]/?o=[NUMID]#job/[NUMID]/run/[NUMID] +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... +Waiting for job run to complete: [DATABRICKS_URL]/?o=[NUMID]#job/[NUMID]/run/[NUMID] +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! diff --git a/acceptance/bundle/resources/job_runs/idempotent_recreate/script b/acceptance/bundle/resources/job_runs/idempotent_recreate/script new file mode 100644 index 00000000000..82750d55a08 --- /dev/null +++ b/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 diff --git a/acceptance/bundle/resources/job_runs/job_parameters/output.txt b/acceptance/bundle/resources/job_runs/job_parameters/output.txt index bcf3f21e017..c2a456016a5 100644 --- a/acceptance/bundle/resources/job_runs/job_parameters/output.txt +++ b/acceptance/bundle/resources/job_runs/job_parameters/output.txt @@ -3,6 +3,7 @@ >>> [CLI] bundle deploy Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-job-parameters/default/files... Deploying resources... +Waiting for job run to complete: [DATABRICKS_URL]/?o=[NUMID]#job/[NUMID]/run/[NUMID] Updating deployment state... Deployment complete! @@ -11,6 +12,7 @@ Deployment complete! "method": "POST", "path": "/api/2.2/jobs/run-now", "body": { + "idempotency_token": "[IDEMPOTENCY_TOKEN]", "job_id": [NUMID], "job_parameters": { "env": "prod" diff --git a/acceptance/bundle/resources/job_runs/job_parameters/test.toml b/acceptance/bundle/resources/job_runs/job_parameters/test.toml deleted file mode 100644 index 4b94d8b58e9..00000000000 --- a/acceptance/bundle/resources/job_runs/job_parameters/test.toml +++ /dev/null @@ -1,4 +0,0 @@ -# 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 diff --git a/acceptance/bundle/resources/job_runs/redeploy/output.txt b/acceptance/bundle/resources/job_runs/redeploy/output.txt index 6ec19be8bce..c3c6ee9d355 100644 --- a/acceptance/bundle/resources/job_runs/redeploy/output.txt +++ b/acceptance/bundle/resources/job_runs/redeploy/output.txt @@ -3,6 +3,7 @@ >>> [CLI] bundle deploy Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-redeploy/default/files... Deploying resources... +Waiting for job run to complete: [DATABRICKS_URL]/?o=[NUMID]#job/[MY_JOB_ID]/run/[NUMID] Updating deployment state... Deployment complete! @@ -30,6 +31,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 +69,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": { @@ -84,6 +87,7 @@ Resources: >>> [CLI] bundle deploy Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-redeploy/default/files... Deploying resources... +Waiting for job run to complete: [DATABRICKS_URL]/?o=[NUMID]#job/[MY_JOB_ID]/run/[NUMID] Updating deployment state... Deployment complete! @@ -103,21 +107,15 @@ Resources: Name: my-job URL: [DATABRICKS_URL]/jobs/[MY_JOB_ID]?w=[NUMID] -=== the config change deleted the previous run and triggered a second, different run +=== the config change triggered a second, different run; the previous run is left in place >>> print_requests.py --keep //jobs/runs/delete -{ - "method": "POST", - "path": "/api/2.2/jobs/runs/delete", - "body": { - "run_id": [NUMID] - } -} >>> print_requests.py //jobs/run-now { "method": "POST", "path": "/api/2.2/jobs/run-now", "body": { + "idempotency_token": "[IDEMPOTENCY_TOKEN]", "job_id": [MY_JOB_ID], "job_parameters": { "env": "prod" diff --git a/acceptance/bundle/resources/job_runs/redeploy/script b/acceptance/bundle/resources/job_runs/redeploy/script index 87d035b9eee..1eed52ca460 100644 --- a/acceptance/bundle/resources/job_runs/redeploy/script +++ b/acceptance/bundle/resources/job_runs/redeploy/script @@ -16,6 +16,8 @@ trace $CLI bundle plan -o json | jq '.plan["resources.job_runs.my_run"]' trace $CLI bundle deploy trace $CLI bundle summary -title "the config change deleted the previous run and triggered a second, different run" +title "the config change triggered a second, different run; the previous run is left in place" +# recreate leaves the old run untouched (no jobs/runs/delete) and fires a fresh +# run-now with the new parameters. trace print_requests.py --keep //jobs/runs/delete trace print_requests.py //jobs/run-now diff --git a/acceptance/bundle/resources/job_runs/redeploy/test.toml b/acceptance/bundle/resources/job_runs/redeploy/test.toml deleted file mode 100644 index 4b94d8b58e9..00000000000 --- a/acceptance/bundle/resources/job_runs/redeploy/test.toml +++ /dev/null @@ -1,4 +0,0 @@ -# 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 diff --git a/acceptance/bundle/resources/job_runs/rerun/databricks.yml b/acceptance/bundle/resources/job_runs/rerun/databricks.yml new file mode 100644 index 00000000000..e819bf786bc --- /dev/null +++ b/acceptance/bundle/resources/job_runs/rerun/databricks.yml @@ -0,0 +1,19 @@ +bundle: + name: job-runs-rerun + +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} + # rerun is folded into the computed idempotency_token but never sent to the + # API. Bumping its value is the supported way to force a fresh run of an + # otherwise-unchanged config. + rerun: v1 diff --git a/acceptance/bundle/resources/job_runs/rerun/out.test.toml b/acceptance/bundle/resources/job_runs/rerun/out.test.toml new file mode 100644 index 00000000000..e90b6d5d1ba --- /dev/null +++ b/acceptance/bundle/resources/job_runs/rerun/out.test.toml @@ -0,0 +1,3 @@ +Local = true +Cloud = false +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] diff --git a/acceptance/bundle/resources/job_runs/rerun/output.txt b/acceptance/bundle/resources/job_runs/rerun/output.txt new file mode 100644 index 00000000000..4a4b0577cdd --- /dev/null +++ b/acceptance/bundle/resources/job_runs/rerun/output.txt @@ -0,0 +1,58 @@ + +=== deploy triggers the first run +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-rerun/default/files... +Deploying resources... +Waiting for job run to complete: [DATABRICKS_URL]/?o=[NUMID]#job/[NUMID]/run/[NUMID] +Updating deployment state... +Deployment complete! + +=== redeploy with identical config is a no-op: the run is reused, no new run-now +>>> [CLI] bundle plan -o json +"skip" + +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-rerun/default/files... +Deploying resources... +Updating deployment state... +Deployment complete! + +=== bump rerun v1 -> v2: the config changed, so a fresh, different run is triggered +>>> update_file.py databricks.yml rerun: v1 rerun: v2 + +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-rerun/default/files... +Deploying resources... +Waiting for job run to complete: [DATABRICKS_URL]/?o=[NUMID]#job/[NUMID]/run/[NUMID] +Updating deployment state... +Deployment complete! + +=== run-now was issued exactly twice (initial + bump); rerun itself is never sent to the API +>>> 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] + } +} +identical config reused run [NUMID]; bumping rerun triggered a new run [NUMID] + +>>> [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-rerun/default + +Deleting files... +Destroy complete! diff --git a/acceptance/bundle/resources/job_runs/rerun/script b/acceptance/bundle/resources/job_runs/rerun/script new file mode 100644 index 00000000000..12f4931391a --- /dev/null +++ b/acceptance/bundle/resources/job_runs/rerun/script @@ -0,0 +1,35 @@ +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 "deploy triggers the first run" +trace $CLI bundle deploy +first=$(run_id) + +title "redeploy with identical config is a no-op: the run is reused, no new run-now" +# rerun is unchanged, so the token is unchanged and the planner skips the run. +trace $CLI bundle plan -o json | jq '.plan["resources.job_runs.my_run"].action // "none"' +trace $CLI bundle deploy +same=$(run_id) + +title "bump rerun v1 -> v2: the config changed, so a fresh, different run is triggered" +trace update_file.py databricks.yml "rerun: v1" "rerun: v2" +trace $CLI bundle deploy +bumped=$(run_id) + +title "run-now was issued exactly twice (initial + bump); rerun itself is never sent to the API" +# Two run-now calls total; the identical-config redeploy in between added none. +# Neither body carries `rerun` -- it only feeds the masked idempotency_token. +trace print_requests.py //jobs/run-now +if [ "$first" = "$same" ] && [ "$first" != "$bumped" ]; then + echo "identical config reused run $first; bumping rerun triggered a new run $bumped" +else + echo "unexpected run ids: first=$first same=$same bumped=$bumped" +fi diff --git a/acceptance/bundle/resources/job_runs/test.toml b/acceptance/bundle/resources/job_runs/test.toml new file mode 100644 index 00000000000..70162af7a72 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/test.toml @@ -0,0 +1,10 @@ +# 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 that differs every run (job_id comes from a +# time-based counter). Mask just that field so run-now goldens stay stable. +[[Repls]] +Old = '"idempotency_token": "[0-9a-f]{64}"' +New = '"idempotency_token": "[IDEMPOTENCY_TOKEN]"' diff --git a/acceptance/bundle/resources/job_runs/wait_output/databricks.yml b/acceptance/bundle/resources/job_runs/wait_output/databricks.yml new file mode 100644 index 00000000000..a694ea66492 --- /dev/null +++ b/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} diff --git a/acceptance/bundle/resources/job_runs/wait_output/out.test.toml b/acceptance/bundle/resources/job_runs/wait_output/out.test.toml new file mode 100644 index 00000000000..e90b6d5d1ba --- /dev/null +++ b/acceptance/bundle/resources/job_runs/wait_output/out.test.toml @@ -0,0 +1,3 @@ +Local = true +Cloud = false +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] diff --git a/acceptance/bundle/resources/job_runs/wait_output/output.txt b/acceptance/bundle/resources/job_runs/wait_output/output.txt new file mode 100644 index 00000000000..82bbf90e1d7 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/wait_output/output.txt @@ -0,0 +1,85 @@ + +=== 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... +Waiting for job run to complete: [DATABRICKS_URL]/?o=[NUMID]#job/[NUMID]/run/[NUMID] +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! diff --git a/acceptance/bundle/resources/job_runs/wait_output/script b/acceptance/bundle/resources/job_runs/wait_output/script new file mode 100644 index 00000000000..a03e99c6777 --- /dev/null +++ b/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 DoCreate 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 diff --git a/bundle/config/resources/job_run.go b/bundle/config/resources/job_run.go index 8db6ced76e1..b67b4c9a561 100644 --- a/bundle/config/resources/job_run.go +++ b/bundle/config/resources/job_run.go @@ -20,6 +20,11 @@ type JobRun struct { BaseResource jobs.RunNow + // Rerun forces a new run when its value changes; it is folded into the + // computed idempotency_token but never sent to the API. Leave it unset for + // normal deploys; bump it to re-run identical config or retry a failed run. + Rerun string `json:"rerun,omitempty"` + // ResolvedJobID holds the run's job_id loaded from state, used only to build // the run URL. Keeping it separate from RunNow.JobId (a ${resources.jobs.*.id} // reference) lets state loading preserve that reference and its plan dependency. diff --git a/bundle/config/validate/validate_job_run_idempotency_token.go b/bundle/config/validate/validate_job_run_idempotency_token.go new file mode 100644 index 00000000000..f198d51c4ce --- /dev/null +++ b/bundle/config/validate/validate_job_run_idempotency_token.go @@ -0,0 +1,49 @@ +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 in DoCreate, so a user value has no effect. + // Reject it and point at the supported knob (`rerun`) for forcing a run. + for name, jr := range b.Config.Resources.JobRuns { + // An empty `job_runs.:` entry loads as a present key with a nil + // pointer value, so guard before dereferencing. + if jr == nil || 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; set `rerun` to force a new run", + 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 +} diff --git a/bundle/config/validate/validate_job_run_idempotency_token_test.go b/bundle/config/validate/validate_job_run_idempotency_token_test.go new file mode 100644 index 00000000000..8f4a36ccb33 --- /dev/null +++ b/bundle/config/validate/validate_job_run_idempotency_token_test.go @@ -0,0 +1,71 @@ +package validate + +import ( + "testing" + + "github.com/databricks/cli/bundle" + "github.com/databricks/cli/bundle/config" + "github.com/databricks/cli/bundle/config/resources" + "github.com/databricks/cli/libs/diag" + "github.com/databricks/databricks-sdk-go/service/jobs" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func jobRunBundle(token string) *bundle.Bundle { + return &bundle.Bundle{ + Config: config.Root{ + Resources: config.Resources{ + JobRuns: map[string]*resources.JobRun{ + "my_run": {RunNow: jobs.RunNow{JobId: 1, IdempotencyToken: token}}, + }, + }, + }, + } +} + +func TestValidateJobRunIdempotencyTokenRejectsUserValue(t *testing.T) { + diags := ValidateJobRunIdempotencyToken().Apply(t.Context(), jobRunBundle("x")) + require.Len(t, diags, 1) + assert.Equal(t, diag.Error, diags[0].Severity) + assert.Equal(t, "idempotency_token is computed automatically and must not be set in bundle configuration; set `rerun` to force a new run", diags[0].Summary) + assert.Equal(t, "resources.job_runs.my_run.idempotency_token", diags[0].Paths[0].String()) +} + +func TestValidateJobRunIdempotencyTokenAllowsUnset(t *testing.T) { + diags := ValidateJobRunIdempotencyToken().Apply(t.Context(), jobRunBundle("")) + require.Empty(t, diags) +} + +func TestValidateJobRunIdempotencyTokenAllowsNilEntry(t *testing.T) { + // An empty `job_runs.:` entry unmarshals to a nil pointer; the + // validator must skip it rather than panic dereferencing. + b := &bundle.Bundle{ + Config: config.Root{ + Resources: config.Resources{ + JobRuns: map[string]*resources.JobRun{"my_run": nil}, + }, + }, + } + diags := ValidateJobRunIdempotencyToken().Apply(t.Context(), b) + require.Empty(t, diags) +} + +func TestValidateJobRunIdempotencyTokenReportsAllSorted(t *testing.T) { + b := &bundle.Bundle{ + Config: config.Root{ + Resources: config.Resources{ + JobRuns: map[string]*resources.JobRun{ + "b_run": {RunNow: jobs.RunNow{JobId: 1, IdempotencyToken: "x"}}, + "a_run": {RunNow: jobs.RunNow{JobId: 2, IdempotencyToken: "y"}}, + }, + }, + }, + } + + diags := ValidateJobRunIdempotencyToken().Apply(t.Context(), b) + require.Len(t, diags, 2) + // Sorted by path, so a_run comes before b_run regardless of map order. + assert.Equal(t, "resources.job_runs.a_run.idempotency_token", diags[0].Paths[0].String()) + assert.Equal(t, "resources.job_runs.b_run.idempotency_token", diags[1].Paths[0].String()) +} diff --git a/bundle/direct/apply.go b/bundle/direct/apply.go index d227d7f564c..dba41cfc9d6 100644 --- a/bundle/direct/apply.go +++ b/bundle/direct/apply.go @@ -27,6 +27,7 @@ func (d *DeploymentUnit) Destroy(ctx context.Context, db *dstate.DeploymentState func (d *DeploymentUnit) Deploy(ctx context.Context, db *dstate.DeploymentState, newState any, actionType deployplan.ActionType, planEntry *deployplan.PlanEntry) error { ctx = log.WithPrefix(ctx, "deploying "+d.ResourceKey) + ctx = dresources.WithResourceIdentity(ctx, d.ResourceKey) if actionType == deployplan.Create { return d.Create(ctx, db, newState) } diff --git a/bundle/direct/dresources/all_test.go b/bundle/direct/dresources/all_test.go index 80e186ca9b8..6bb976e7d4a 100644 --- a/bundle/direct/dresources/all_test.go +++ b/bundle/direct/dresources/all_test.go @@ -1075,7 +1075,9 @@ func testCRUD(t *testing.T, group string, adapter *Adapter, client *databricks.W require.NoError(t, err) } - deleteIsNoop := strings.HasSuffix(group, "permissions") || strings.HasSuffix(group, "grants") + // job_runs, like permissions/grants, has a noop DoDelete: a run is an + // immutable historical record left in place, so DoRead still finds it. + deleteIsNoop := strings.HasSuffix(group, "permissions") || strings.HasSuffix(group, "grants") || group == "job_runs" // Apps DoDelete is fire-and-forget: the API returns success while the app // sits in DELETING state for up to ~20 minutes before the record is removed. // A GET on the DELETING app returns the app, not 404 -- the testserver diff --git a/bundle/direct/dresources/identity.go b/bundle/direct/dresources/identity.go new file mode 100644 index 00000000000..a21ff4d0c64 --- /dev/null +++ b/bundle/direct/dresources/identity.go @@ -0,0 +1,19 @@ +package dresources + +import "context" + +type resourceIdentityKey struct{} + +// WithResourceIdentity records the deployment-scoped key of the resource being +// deployed, so resources can derive a stable per-resource seed (e.g. a run-now +// idempotency token). Set once by the framework in DeploymentUnit.Deploy. +func WithResourceIdentity(ctx context.Context, key string) context.Context { + return context.WithValue(ctx, resourceIdentityKey{}, key) +} + +// ResourceIdentity returns the key set by WithResourceIdentity, or "" if unset +// (e.g. the unit-test harness that calls resources directly). +func ResourceIdentity(ctx context.Context) string { + key, _ := ctx.Value(resourceIdentityKey{}).(string) + return key +} diff --git a/bundle/direct/dresources/job_run.go b/bundle/direct/dresources/job_run.go index 0a2ae0ea6af..d9602b5c702 100644 --- a/bundle/direct/dresources/job_run.go +++ b/bundle/direct/dresources/job_run.go @@ -2,18 +2,32 @@ package dresources import ( "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" "fmt" "strconv" + "time" "github.com/databricks/cli/bundle/config/resources" + "github.com/databricks/cli/libs/cmdio" "github.com/databricks/databricks-sdk-go" "github.com/databricks/databricks-sdk-go/marshal" "github.com/databricks/databricks-sdk-go/service/jobs" ) +// jobRunWaitTimeout bounds the wait so a deploy can't hang forever. Matches +// `bundle run` (jobRunTimeout in bundle/run/job.go). +const jobRunWaitTimeout = 24 * time.Hour + // JobRunState is what we persist for a triggered run: the RunNow request. type JobRunState struct { jobs.RunNow + + // Rerun is a user token folded into the idempotency token (see + // idempotencyToken); changing it recreates the run. Not a RunNow field, so + // never sent to the API. + Rerun string `json:"rerun,omitempty"` } func (s *JobRunState) UnmarshalJSON(b []byte) error { @@ -29,6 +43,11 @@ func (s JobRunState) MarshalJSON() ([]byte, error) { type JobRunRemote struct { jobs.RunNow + // Rerun keeps RemoteType a superset of StateType (TestRemoteSuperset). GetRun + // never returns it, so makeJobRunRemote zeroes it and root + // ignore_remote_changes hides the drift. + Rerun string `json:"rerun,omitempty"` + RunId int64 `json:"run_id,omitempty"` RunName string `json:"run_name,omitempty"` State *jobs.RunState `json:"state,omitempty"` @@ -59,6 +78,7 @@ func (*ResourceJobRun) New(client *databricks.WorkspaceClient) *ResourceJobRun { func (*ResourceJobRun) PrepareState(input *resources.JobRun) *JobRunState { return &JobRunState{ RunNow: input.RunNow, + Rerun: input.Rerun, } } @@ -97,6 +117,8 @@ func makeJobRunRemote(run *jobs.Run) *JobRunRemote { Queue: nil, ForceSendFields: nil, }, + // Bundle-only field GetRun never reports; see the field's doc comment. + Rerun: "", RunId: run.RunId, RunName: run.RunName, State: run.State, @@ -123,33 +145,83 @@ func (r *ResourceJobRun) DoRead(ctx context.Context, id string) (*JobRunRemote, return makeJobRunRemote(run), nil } -// RemapState extracts the embedded RunNow as the state used for diffing. +// RemapState extracts the embedded RunNow as the state used for diffing. Rerun +// is always "" here (bundle-only, never in remote); root ignore_remote_changes +// suppresses the drift against the user's value in state. func (*ResourceJobRun) RemapState(remote *JobRunRemote) *JobRunState { - return &JobRunState{RunNow: remote.RunNow} + return &JobRunState{RunNow: remote.RunNow, Rerun: remote.Rerun} } func (r *ResourceJobRun) DoCreate(ctx context.Context, config *JobRunState) (string, *JobRunRemote, error) { - // RunNow returns only the new run id, so we return a nil remote and let the - // framework read it back via DoRead. - wait, err := r.client.Jobs.RunNow(ctx, config.RunNow) + token, err := idempotencyToken(ctx, config) if err != nil { return "", nil, err } - return strconv.FormatInt(wait.RunId, 10), nil, nil -} + // Set the token on a copy so it reaches the API but never lands in state. + req := config.RunNow + req.IdempotencyToken = token -// DoUpdate is intentionally not implemented: a run can't be modified in place, -// so any change recreates it (delete + a fresh RunNow). + wait, err := r.client.Jobs.RunNow(ctx, req) + if err != nil { + return "", nil, err + } -// DoDelete deletes the run via jobs/runs/delete, on both destroy and the -// recreate path. The API rejects a still-active run; this milestone doesn't -// await completion, so that error surfaces to the user. -func (r *ResourceJobRun) DoDelete(ctx context.Context, id string, _ *JobRunState) error { - runID, err := parseRunID(id) + // Wait here, not in WaitAfterCreate: state persists only after DoCreate + // returns, so an interrupted wait saves nothing and re-triggers RunNow, which + // the idempotency_token rejoins to the same run. Waiting after state is saved + // would let the planner skip the wait on resume. + remote, err := r.waitForRun(ctx, wait.RunId) if err != nil { - return err + return "", nil, err } - return r.client.Jobs.DeleteRunByRunId(ctx, runID) + return strconv.FormatInt(wait.RunId, 10), remote, nil +} + +// waitForRun blocks until the run reaches a terminal state and returns its +// remote view. Only SUCCESS completes the deploy; any other terminal outcome is +// an error. +func (r *ResourceJobRun) waitForRun(ctx context.Context, runID int64) (*JobRunRemote, error) { + // Log the run's UI link on the first poll: the wait can last up to + // jobRunWaitTimeout, so a silent deploy would look hung. + logged := false + run, err := r.client.Jobs.WaitGetRunJobTerminatedOrSkipped(ctx, runID, jobRunWaitTimeout, func(run *jobs.Run) { + if logged || run.RunPageUrl == "" { + return + } + logged = true + // The direct-engine unit harness runs without cmdIO, so guard LogString. + if cmdio.HasIO(ctx) { + cmdio.LogString(ctx, "Waiting for job run to complete: "+run.RunPageUrl) + } + }) + if err != nil { + return nil, err + } + // Every non-SUCCESS terminal outcome (FAILED, TIMEDOUT, CANCELED, + // SUCCESS_WITH_FAILURES, SKIPPED, ...) fails the deploy; the waiter already + // errored on INTERNAL_ERROR and timeout above. + if run.State.ResultState != jobs.RunResultStateSuccess { + outcome := string(run.State.ResultState) + if outcome == "" { + // A skipped run has no result_state; report the lifecycle state. + outcome = string(run.State.LifeCycleState) + } + if run.State.StateMessage != "" { + return nil, fmt.Errorf("job run %d did not succeed: %s: %s", runID, outcome, run.State.StateMessage) + } + return nil, fmt.Errorf("job run %d did not succeed: %s", runID, outcome) + } + return makeJobRunRemote(run), nil +} + +// DoUpdate is intentionally not implemented: a run can't be modified in place, +// so any change recreates it (a fresh RunNow; the prior run is left in place). + +// DoDelete is a noop: a run is immutable history, so destroy/recreate leave it +// in place. Deleting would tombstone its idempotency_token (see JobsRunNow) and +// break re-running the same config. +func (*ResourceJobRun) DoDelete(_ context.Context, _ string, _ *JobRunState) error { + return nil } func parseRunID(id string) (int64, error) { @@ -159,3 +231,21 @@ func parseRunID(id string) (int64, error) { } return result, nil } + +// idempotencyToken hashes the resource key plus config into a stable token: the +// key keeps identical configs distinct (both run), any config change (including +// the user-set rerun field) re-triggers, and an unchanged retry reuses the run. +// Hex SHA-256 (64 chars, Jobs API max). +func idempotencyToken(ctx context.Context, state *JobRunState) (string, error) { + toHash := *state + toHash.IdempotencyToken = "" + canonical, err := json.Marshal(toHash) + if err != nil { + return "", err + } + h := sha256.New() + h.Write([]byte(ResourceIdentity(ctx))) + h.Write([]byte{0}) // separator so key‖config can't collide via a shifted boundary + h.Write(canonical) + return hex.EncodeToString(h.Sum(nil)), nil +} diff --git a/bundle/direct/dresources/job_run_test.go b/bundle/direct/dresources/job_run_test.go new file mode 100644 index 00000000000..36e9ec9183d --- /dev/null +++ b/bundle/direct/dresources/job_run_test.go @@ -0,0 +1,187 @@ +package dresources + +import ( + "sync/atomic" + "testing" + + "github.com/databricks/cli/libs/testserver" + "github.com/databricks/databricks-sdk-go" + "github.com/databricks/databricks-sdk-go/service/jobs" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestIdempotencyTokenIsStableHex(t *testing.T) { + ctx := t.Context() + run := jobs.RunNow{JobId: 123, JobParameters: map[string]string{"env": "prod"}} + + got, err := idempotencyToken(ctx, &JobRunState{RunNow: run}) + require.NoError(t, err) + + // hex SHA-256 is always 64 lowercase hex chars (the Jobs API maximum). + assert.Regexp(t, "^[0-9a-f]{64}$", got) + + // Deterministic: the same config yields the same token, so a retry dedupes. + again, err := idempotencyToken(ctx, &JobRunState{RunNow: run}) + require.NoError(t, err) + assert.Equal(t, got, again) +} + +func TestIdempotencyTokenIgnoresPresetToken(t *testing.T) { + ctx := t.Context() + a, err := idempotencyToken(ctx, &JobRunState{RunNow: jobs.RunNow{JobId: 123}}) + require.NoError(t, err) + b, err := idempotencyToken(ctx, &JobRunState{RunNow: jobs.RunNow{JobId: 123, IdempotencyToken: "user-supplied"}}) + require.NoError(t, err) + + // The token is cleared before hashing, so a preset value cannot change it. + assert.Equal(t, a, b) +} + +func TestIdempotencyTokenChangesWithConfig(t *testing.T) { + ctx := t.Context() + dev, err := idempotencyToken(ctx, &JobRunState{RunNow: jobs.RunNow{JobId: 123, JobParameters: map[string]string{"env": "dev"}}}) + require.NoError(t, err) + prod, err := idempotencyToken(ctx, &JobRunState{RunNow: jobs.RunNow{JobId: 123, JobParameters: map[string]string{"env": "prod"}}}) + require.NoError(t, err) + otherJob, err := idempotencyToken(ctx, &JobRunState{RunNow: jobs.RunNow{JobId: 456}}) + require.NoError(t, err) + + assert.NotEqual(t, dev, prod) // different params --> different token + assert.NotEqual(t, dev, otherJob) // different job_id --> different token +} + +func TestIdempotencyTokenChangesWithRerun(t *testing.T) { + ctx := t.Context() + run := jobs.RunNow{JobId: 123} + + base, err := idempotencyToken(ctx, &JobRunState{RunNow: run}) + require.NoError(t, err) + bumped, err := idempotencyToken(ctx, &JobRunState{RunNow: run, Rerun: "v2"}) + require.NoError(t, err) + bumpedAgain, err := idempotencyToken(ctx, &JobRunState{RunNow: run, Rerun: "v2"}) + require.NoError(t, err) + + assert.NotEqual(t, base, bumped) // bumping rerun forces a new run + assert.Equal(t, bumped, bumpedAgain) // the same rerun value stays stable +} + +func TestIdempotencyTokenChangesWithResourceIdentity(t *testing.T) { + run := jobs.RunNow{JobId: 123} + a, err := idempotencyToken(WithResourceIdentity(t.Context(), "resources.job_runs.a"), &JobRunState{RunNow: run}) + require.NoError(t, err) + b, err := idempotencyToken(WithResourceIdentity(t.Context(), "resources.job_runs.b"), &JobRunState{RunNow: run}) + require.NoError(t, err) + + // Identical config under different resource keys must not collapse onto one run. + assert.NotEqual(t, a, b) +} + +// jobRunClient returns a client whose GetRun always reports the given terminal +// run state, so waitForRun can be exercised without a real run. +func jobRunClient(t *testing.T, state *jobs.RunState) *databricks.WorkspaceClient { + t.Helper() + server := testserver.New(t) + // First registration wins, so this overrides the default runs/get handler. + server.Handle("GET", "/api/2.2/jobs/runs/get", func(req testserver.Request) any { + return jobs.Run{RunId: 123, JobId: 456, State: state} + }) + testserver.AddDefaultHandlers(server) + + client, err := databricks.NewWorkspaceClient(&databricks.Config{ + Host: server.URL, + Token: "testtoken", + }) + require.NoError(t, err) + return client +} + +func TestJobRunWaitFailsOnFailedResult(t *testing.T) { + client := jobRunClient(t, &jobs.RunState{ + LifeCycleState: jobs.RunLifeCycleStateTerminated, + ResultState: jobs.RunResultStateFailed, + StateMessage: "task failed", + }) + + r := (&ResourceJobRun{}).New(client) + _, err := r.waitForRun(t.Context(), 123) + + // Only SUCCESS completes the deploy; a FAILED result fails it. + require.ErrorContains(t, err, "did not succeed: FAILED: task failed") +} + +func TestJobRunWaitFailsOnSkipped(t *testing.T) { + // A skipped run has no result_state, so the lifecycle state is reported. + client := jobRunClient(t, &jobs.RunState{ + LifeCycleState: jobs.RunLifeCycleStateSkipped, + }) + + r := (&ResourceJobRun{}).New(client) + _, err := r.waitForRun(t.Context(), 123) + + require.ErrorContains(t, err, "did not succeed: SKIPPED") +} + +func TestJobRunWaitSucceeds(t *testing.T) { + client := jobRunClient(t, &jobs.RunState{ + LifeCycleState: jobs.RunLifeCycleStateTerminated, + ResultState: jobs.RunResultStateSuccess, + }) + + r := (&ResourceJobRun{}).New(client) + remote, err := r.waitForRun(t.Context(), 123) + + require.NoError(t, err) + require.NotNil(t, remote.State) + assert.Equal(t, jobs.RunResultStateSuccess, remote.State.ResultState) +} + +func TestJobRunWaitFailsOnInternalError(t *testing.T) { + client := jobRunClient(t, &jobs.RunState{ + LifeCycleState: jobs.RunLifeCycleStateInternalError, + }) + + r := (&ResourceJobRun{}).New(client) + _, err := r.waitForRun(t.Context(), 123) + + // INTERNAL_ERROR is the one terminal state that fails the deploy. + require.Error(t, err) +} + +// Reports RUNNING before TERMINATED, so this is the only test that exercises the +// poll loop (the others stub an already-terminal state). +func TestJobRunWaitPollsUntilTerminal(t *testing.T) { + server := testserver.New(t) + + // waitForRun's progress callback fires on every poll (there is no separate + // GET before the loop), so returning RUNNING for the first two polls just + // makes the waiter iterate more than once before it sees TERMINATED. + var gets atomic.Int32 + server.Handle("GET", "/api/2.2/jobs/runs/get", func(req testserver.Request) any { + if gets.Add(1) <= 2 { + return jobs.Run{RunId: 123, JobId: 456, State: &jobs.RunState{ + LifeCycleState: jobs.RunLifeCycleStateRunning, + }} + } + return jobs.Run{RunId: 123, JobId: 456, State: &jobs.RunState{ + LifeCycleState: jobs.RunLifeCycleStateTerminated, + ResultState: jobs.RunResultStateSuccess, + }} + }) + testserver.AddDefaultHandlers(server) + + client, err := databricks.NewWorkspaceClient(&databricks.Config{ + Host: server.URL, + Token: "testtoken", + }) + require.NoError(t, err) + + r := (&ResourceJobRun{}).New(client) + remote, err := r.waitForRun(t.Context(), 123) + require.NoError(t, err) + + // SUCCESS is only reachable by polling past the RUNNING reads. + require.NotNil(t, remote.State) + assert.Equal(t, jobs.RunResultStateSuccess, remote.State.ResultState) + assert.GreaterOrEqual(t, gets.Load(), int32(2), "expected waitForRun to poll more than once") +} diff --git a/bundle/internal/schema/annotations.yml b/bundle/internal/schema/annotations.yml index 10832fe04a6..9748abc00e8 100644 --- a/bundle/internal/schema/annotations.yml +++ b/bundle/internal/schema/annotations.yml @@ -975,6 +975,9 @@ resources: "python_named_params": "description": |- PLACEHOLDER + "rerun": + "description": |- + An arbitrary token that forces a new run when its value changes. Leave it unset for normal deploys; change it to re-run the same job configuration or to retry a run that finished unsuccessfully. "jobs": "description": |- The job definitions for the bundle, where each key is the name of the job. diff --git a/bundle/phases/initialize.go b/bundle/phases/initialize.go index bfa2af4124b..56b7a6138f7 100644 --- a/bundle/phases/initialize.go +++ b/bundle/phases/initialize.go @@ -177,6 +177,9 @@ func Initialize(ctx context.Context, b *bundle.Bundle) { // They are set by the CLI to track the bundle deployment and must not be set by the user. validate.ValidateDeploymentFields(), + // job_runs' idempotency_token is computed automatically, so reject a user-set value. + validate.ValidateJobRunIdempotencyToken(), + // Reads (dynamic): * (strings) (searches for ${resources.*} references) // Warns (TF engine) or errors (direct engine) when a cross-resource reference // points to a Terraform-only field with no DABs equivalent. diff --git a/bundle/schema/jsonschema.json b/bundle/schema/jsonschema.json index 4c78bd7c384..8c647c01af1 100644 --- a/bundle/schema/jsonschema.json +++ b/bundle/schema/jsonschema.json @@ -1232,6 +1232,10 @@ "description": "The queue settings of the run.", "$ref": "#/$defs/github.com/databricks/databricks-sdk-go/service/jobs.QueueSettings" }, + "rerun": { + "description": "An arbitrary token that forces a new run when its value changes. Leave it unset for normal deploys; change it to re-run the same job configuration or to retry a run that finished unsuccessfully.", + "$ref": "#/$defs/string" + }, "spark_submit_params": { "description": "[Private Preview] A list of parameters for jobs with spark submit task, for example `\"spark_submit_params\": [\"--class\", \"org.apache.spark.examples.SparkPi\"]`.\nThe parameters are passed to spark-submit script as command-line parameters. If specified upon `run-now`, it would overwrite the\nparameters specified in job setting. The JSON representation of this field (for example `{\"python_params\":[\"john doe\",\"35\"]}`)\ncannot exceed 10,000 bytes.\n\n⚠ **Deprecation note** Use [job parameters](https://docs.databricks.com/jobs/job-parameters.html#job-parameter-pushdown) to pass information down to tasks.\n\nImportant\n\nThese parameters accept only Latin characters (ASCII character set). Using non-ASCII characters returns an error.\nExamples of invalid, non-ASCII characters are Chinese, Japanese kanjis, and emojis.", "$ref": "#/$defs/slice/string", diff --git a/libs/testserver/fake_workspace.go b/libs/testserver/fake_workspace.go index 8d6e8ee0dd3..cf7d928a9ae 100644 --- a/libs/testserver/fake_workspace.go +++ b/libs/testserver/fake_workspace.go @@ -169,6 +169,7 @@ type FakeWorkspace struct { Jobs map[int64]jobs.Job JobRuns map[int64]jobs.Run JobRunOutputs map[int64]jobs.RunOutput + JobRunsByToken map[string]int64 Pipelines map[string]pipelines.GetPipelineResponse PipelineUpdates map[string]bool Monitors map[string]catalog.MonitorInfo @@ -335,6 +336,7 @@ func NewFakeWorkspace(url, token string) *FakeWorkspace { Jobs: map[int64]jobs.Job{}, JobRuns: map[int64]jobs.Run{}, JobRunOutputs: map[int64]jobs.RunOutput{}, + JobRunsByToken: map[string]int64{}, Grants: map[string][]catalog.PrivilegeAssignment{}, Pipelines: map[string]pipelines.GetPipelineResponse{}, PipelineUpdates: map[string]bool{}, diff --git a/libs/testserver/jobs.go b/libs/testserver/jobs.go index 9097d43c086..cf0ba0b03f7 100644 --- a/libs/testserver/jobs.go +++ b/libs/testserver/jobs.go @@ -349,6 +349,22 @@ func (s *FakeWorkspace) JobsRunNow(req Request) Response { return Response{StatusCode: 404} } + // run-now is idempotent: the same token returns the existing run. The token + // is not freed when its run is deleted, so reusing it then errors instead of + // starting a fresh run. + // https://docs.databricks.com/api/workspace/jobs/runnow + if request.IdempotencyToken != "" { + if existing, ok := s.JobRunsByToken[request.IdempotencyToken]; ok { + if _, alive := s.JobRuns[existing]; alive { + return Response{Body: jobs.RunNowResponse{RunId: existing}} + } + return Response{ + StatusCode: 400, + Body: fmt.Sprintf("idempotency_token %q was used for run %d, which has been deleted", request.IdempotencyToken, existing), + } + } + } + runId := nextID() runName := "run-name" if job.Settings != nil && job.Settings.Name != "" { @@ -412,6 +428,10 @@ func (s *FakeWorkspace) JobsRunNow(req Request) Response { OverridingParameters: runOverridingParameters(request), } + if request.IdempotencyToken != "" { + s.JobRunsByToken[request.IdempotencyToken] = runId + } + return Response{Body: jobs.RunNowResponse{RunId: runId}} } @@ -899,7 +919,19 @@ func (s *FakeWorkspace) JobsDeleteRun(req Request) Response { Body: fmt.Sprintf("request parsing error: %s", err), } } - return MapDelete(s, s.JobRuns, request.RunId) + + defer s.LockUnlock()() + + if _, ok := s.JobRuns[request.RunId]; !ok { + return Response{StatusCode: 404} + } + delete(s.JobRuns, request.RunId) + + // Keep the token->run mapping as a tombstone: the Jobs API does not free an + // idempotency_token when its run is deleted, so a later reuse errors (see + // JobsRunNow) rather than starting a fresh run. + + return Response{} } func (s *FakeWorkspace) JobsGetRunOutput(req Request) Response { diff --git a/libs/testserver/jobs_test.go b/libs/testserver/jobs_test.go index 38e47b68b08..05b9b24b382 100644 --- a/libs/testserver/jobs_test.go +++ b/libs/testserver/jobs_test.go @@ -87,6 +87,54 @@ func TestJobsSubmit_RunReachesTerminalStateOnPoll(t *testing.T) { assert.Equal(t, jobs.RunResultStateSuccess, second.State.ResultState) } +func createJob(t *testing.T, workspace *FakeWorkspace) int64 { + t.Helper() + body, err := json.Marshal(jobs.CreateJob{Name: "my-job"}) + require.NoError(t, err) + + response := workspace.JobsCreate(Request{Body: body}) + require.Equal(t, 0, response.StatusCode) + return response.Body.(jobs.CreateResponse).JobId +} + +func runNow(t *testing.T, workspace *FakeWorkspace, request jobs.RunNow) Response { + t.Helper() + body, err := json.Marshal(request) + require.NoError(t, err) + return workspace.JobsRunNow(Request{Body: body}) +} + +// The Jobs API does not free an idempotency_token when its run is deleted, so a +// later reuse errors instead of starting a fresh run. This is exactly why +// job_runs' DoDelete is a noop: deleting a run would tombstone the deterministic +// token and break re-running the same config after destroy or a value flip-back. +func TestJobsRunNow_IdempotencyTokenTombstonedAfterDelete(t *testing.T) { + workspace := NewFakeWorkspace("http://test", "dbapi123") + jobID := createJob(t, workspace) + + const token = "stable-token" + + first := runNow(t, workspace, jobs.RunNow{JobId: jobID, IdempotencyToken: token}) + require.Equal(t, 0, first.StatusCode) + runID := first.Body.(jobs.RunNowResponse).RunId + require.NotZero(t, runID) + + // A retry with the same token dedupes to the existing run (no duplicate). + second := runNow(t, workspace, jobs.RunNow{JobId: jobID, IdempotencyToken: token}) + require.Equal(t, 0, second.StatusCode) + assert.Equal(t, runID, second.Body.(jobs.RunNowResponse).RunId) + + body, err := json.Marshal(jobs.DeleteRun{RunId: runID}) + require.NoError(t, err) + del := workspace.JobsDeleteRun(Request{Body: body}) + require.Equal(t, 0, del.StatusCode) + + // Reusing the tombstoned token now errors rather than starting a fresh run. + third := runNow(t, workspace, jobs.RunNow{JobId: jobID, IdempotencyToken: token}) + assert.Equal(t, 400, third.StatusCode) + assert.Contains(t, third.Body.(string), "has been deleted") +} + func TestJobsSubmit_RejectsInvalidGitProvider(t *testing.T) { workspace := NewFakeWorkspace("http://test", "dbapi123")