From c8f93ddd550ce7b76c4c90b79053ba9738bbd7aa Mon Sep 17 00:00:00 2001 From: Rada Kamysheva Date: Thu, 9 Jul 2026 14:17:18 +0000 Subject: [PATCH 1/6] Add idempotency_token to job_runs resource Job runs now send an automatic idempotency_token on run-now so a retried deployment reuses the existing run instead of triggering a duplicate. The token is the hex SHA-256 of the RunNow request (job_id cleared of the token itself), computed in DoCreate on a copy so it is never persisted to state. A user-set idempotency_token is rejected at validation time, and the local testserver now deduplicates run-now by token so the behavior is covered by tests. --- .../resources/job_runs/basic/output.txt | 1 + .../bundle/resources/job_runs/basic/test.toml | 4 - .../job_runs/job_parameters/output.txt | 1 + .../job_runs/job_parameters/test.toml | 4 - .../resources/job_runs/redeploy/output.txt | 2 + .../resources/job_runs/redeploy/test.toml | 4 - .../bundle/resources/job_runs/test.toml | 11 +++ .../validate_job_run_idempotency_token.go | 48 ++++++++++++ ...validate_job_run_idempotency_token_test.go | 57 ++++++++++++++ bundle/direct/dresources/job_run.go | 28 ++++++- bundle/direct/dresources/job_run_test.go | 78 +++++++++++++++++++ bundle/phases/initialize.go | 3 + libs/testserver/fake_workspace.go | 2 + libs/testserver/jobs.go | 12 +++ 14 files changed, 242 insertions(+), 13 deletions(-) delete mode 100644 acceptance/bundle/resources/job_runs/basic/test.toml delete mode 100644 acceptance/bundle/resources/job_runs/job_parameters/test.toml delete mode 100644 acceptance/bundle/resources/job_runs/redeploy/test.toml create mode 100644 acceptance/bundle/resources/job_runs/test.toml create mode 100644 bundle/config/validate/validate_job_run_idempotency_token.go create mode 100644 bundle/config/validate/validate_job_run_idempotency_token_test.go create mode 100644 bundle/direct/dresources/job_run_test.go diff --git a/acceptance/bundle/resources/job_runs/basic/output.txt b/acceptance/bundle/resources/job_runs/basic/output.txt index 14018b93ec6..b5872e724b0 100644 --- a/acceptance/bundle/resources/job_runs/basic/output.txt +++ b/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] } } 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/job_parameters/output.txt b/acceptance/bundle/resources/job_runs/job_parameters/output.txt index bcf3f21e017..5d4761bf2b6 100644 --- a/acceptance/bundle/resources/job_runs/job_parameters/output.txt +++ b/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" 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..b39382c0a4a 100644 --- a/acceptance/bundle/resources/job_runs/redeploy/output.txt +++ b/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" @@ -118,6 +119,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" 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/test.toml b/acceptance/bundle/resources/job_runs/test.toml new file mode 100644 index 00000000000..0a29818c6e5 --- /dev/null +++ b/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]' 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..55ea913827e --- /dev/null +++ b/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 +} 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..1bd8fddeb90 --- /dev/null +++ b/bundle/config/validate/validate_job_run_idempotency_token_test.go @@ -0,0 +1,57 @@ +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", 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 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/dresources/job_run.go b/bundle/direct/dresources/job_run.go index 0a2ae0ea6af..d1ca0659153 100644 --- a/bundle/direct/dresources/job_run.go +++ b/bundle/direct/dresources/job_run.go @@ -2,6 +2,9 @@ package dresources import ( "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" "fmt" "strconv" @@ -129,9 +132,18 @@ func (*ResourceJobRun) RemapState(remote *JobRunRemote) *JobRunState { } func (r *ResourceJobRun) DoCreate(ctx context.Context, config *JobRunState) (string, *JobRunRemote, error) { + // Copy the request so the derived token is sent to the API but never + // persisted into state (state stays token-free, plans stay clean). + req := config.RunNow + token, err := idempotencyToken(req) + if err != nil { + return "", nil, err + } + req.IdempotencyToken = token + // 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) + wait, err := r.client.Jobs.RunNow(ctx, req) if err != nil { return "", nil, err } @@ -159,3 +171,17 @@ func parseRunID(id string) (int64, error) { } return result, nil } + +// idempotencyToken derives a stable token from the RunNow request so that a +// retried run-now returns the existing run instead of creating a duplicate. +// The token is the hex SHA-256 of the request's JSON with idempotency_token +// cleared; hex SHA-256 is exactly 64 chars, the Jobs API maximum. +func idempotencyToken(req jobs.RunNow) (string, error) { + req.IdempotencyToken = "" + canonical, err := json.Marshal(req) + if err != nil { + return "", err + } + sum := sha256.Sum256(canonical) + return hex.EncodeToString(sum[:]), 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..fd1ded7000a --- /dev/null +++ b/bundle/direct/dresources/job_run_test.go @@ -0,0 +1,78 @@ +package dresources + +import ( + "testing" + + "github.com/databricks/databricks-sdk-go/service/jobs" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestIdempotencyTokenIsStableHex(t *testing.T) { + run := jobs.RunNow{JobId: 123, JobParameters: map[string]string{"env": "prod"}} + + got, err := idempotencyToken(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(run) + require.NoError(t, err) + assert.Equal(t, got, again) +} + +func TestIdempotencyTokenIgnoresPresetToken(t *testing.T) { + a, err := idempotencyToken(jobs.RunNow{JobId: 123}) + require.NoError(t, err) + b, err := idempotencyToken(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) { + dev, err := idempotencyToken(jobs.RunNow{JobId: 123, JobParameters: map[string]string{"env": "dev"}}) + require.NoError(t, err) + prod, err := idempotencyToken(jobs.RunNow{JobId: 123, JobParameters: map[string]string{"env": "prod"}}) + require.NoError(t, err) + otherJob, err := idempotencyToken(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 TestJobRunIdempotentCreate(t *testing.T) { + _, client := setupTestServerClient(t) + ctx := t.Context() + + // A run can only target an existing job, so create one first. + job, err := client.Jobs.Create(ctx, jobs.CreateJob{ + Name: "idempotency-job", + Tasks: []jobs.Task{{ + TaskKey: "t", + NotebookTask: &jobs.NotebookTask{NotebookPath: "/Workspace/Users/user@example.com/notebook"}, + }}, + }) + require.NoError(t, err) + + r := (&ResourceJobRun{}).New(client) + config := &JobRunState{RunNow: jobs.RunNow{JobId: job.JobId}} + + // Same config twice: the derived token is identical, so the backend returns + // the existing run instead of creating a duplicate. + id1, _, err := r.DoCreate(ctx, config) + require.NoError(t, err) + id2, _, err := r.DoCreate(ctx, config) + require.NoError(t, err) + assert.Equal(t, id1, id2, "same config must dedupe to the same run") + + // Different config: different token, so a genuinely new run is created. + other := &JobRunState{RunNow: jobs.RunNow{JobId: job.JobId, JobParameters: map[string]string{"env": "prod"}}} + id3, _, err := r.DoCreate(ctx, other) + require.NoError(t, err) + assert.NotEqual(t, id1, id3, "different config must create a new run") +} diff --git a/bundle/phases/initialize.go b/bundle/phases/initialize.go index bfa2af4124b..8d51491b30f 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(), + // Validate that idempotency_token is not set on job runs. It is computed automatically and must not be set by the user. + 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/libs/testserver/fake_workspace.go b/libs/testserver/fake_workspace.go index 1c5d842f0f8..ec0854d3baa 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 @@ -334,6 +335,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 2d019c3e496..b6bb3fa967b 100644 --- a/libs/testserver/jobs.go +++ b/libs/testserver/jobs.go @@ -349,6 +349,14 @@ func (s *FakeWorkspace) JobsRunNow(req Request) Response { return Response{StatusCode: 404} } + // The Jobs API treats run-now as idempotent: the same idempotency_token + // returns the run already created for it instead of starting a new one. + if request.IdempotencyToken != "" { + if existing, ok := s.JobRunsByToken[request.IdempotencyToken]; ok { + return Response{Body: jobs.RunNowResponse{RunId: existing}} + } + } + runId := nextID() runName := "run-name" if job.Settings != nil && job.Settings.Name != "" { @@ -412,6 +420,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}} } From 4285052219754b37b33513b6d11125e1f411e114 Mon Sep 17 00:00:00 2001 From: Rada Kamysheva Date: Fri, 10 Jul 2026 09:30:38 +0000 Subject: [PATCH 2/6] Add acceptance test for job_runs idempotent recreate Covers the idempotency token end-to-end: deploy triggers a run, the run is dropped from local state (keeping the job so job_id and the derived token stay stable), and a redeploy re-issues run-now. The backend dedupes on the token and returns the existing run, so no duplicate is created. Drop the DoCreate-level TestJobRunIdempotentCreate since the acceptance test now covers that path end-to-end; the token helper's unit tests remain. --- .../idempotent_recreate/databricks.yml | 15 +++++ .../idempotent_recreate/out.test.toml | 3 + .../job_runs/idempotent_recreate/output.txt | 61 +++++++++++++++++++ .../job_runs/idempotent_recreate/script | 33 ++++++++++ bundle/direct/dresources/job_run_test.go | 32 ---------- 5 files changed, 112 insertions(+), 32 deletions(-) create mode 100644 acceptance/bundle/resources/job_runs/idempotent_recreate/databricks.yml create mode 100644 acceptance/bundle/resources/job_runs/idempotent_recreate/out.test.toml create mode 100644 acceptance/bundle/resources/job_runs/idempotent_recreate/output.txt create mode 100644 acceptance/bundle/resources/job_runs/idempotent_recreate/script 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..d506c75a435 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/idempotent_recreate/output.txt @@ -0,0 +1,61 @@ + +=== initial deploy triggers the run +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-idempotent-recreate/default/files... +Deploying resources... +Updating deployment state... +Deployment complete! + +=== drop the job run from local state but keep the job (simulates a lost run id) +>>> [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] + } + } +} + +=== redeploy re-issues run-now; the token dedupes to the existing run +>>> [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 twice with the same token, but 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 unchanged after recreate: the idempotency token deduped to 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..e25372521fa --- /dev/null +++ b/acceptance/bundle/resources/job_runs/idempotent_recreate/script @@ -0,0 +1,33 @@ +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" +trace $CLI bundle deploy +before=$(run_id) + +title "drop the job run from local state but keep the job (simulates a lost run id)" +# The job stays deployed, so job_id -- and therefore the derived idempotency +# token -- is unchanged, which is what lets the retried run-now dedupe. +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"]' + +title "redeploy re-issues run-now; the token dedupes to the existing run" +trace $CLI bundle deploy +after=$(run_id) + +title "run-now was issued twice with the same token, but no duplicate run was created" +trace print_requests.py //jobs/run-now +if [ "$before" = "$after" ]; then + echo "run id unchanged after recreate: the idempotency token deduped to the existing run" +else + echo "run id changed ($before -> $after): a duplicate run was created" +fi diff --git a/bundle/direct/dresources/job_run_test.go b/bundle/direct/dresources/job_run_test.go index fd1ded7000a..8c10d8df85f 100644 --- a/bundle/direct/dresources/job_run_test.go +++ b/bundle/direct/dresources/job_run_test.go @@ -44,35 +44,3 @@ func TestIdempotencyTokenChangesWithConfig(t *testing.T) { assert.NotEqual(t, dev, prod) // different params --> different token assert.NotEqual(t, dev, otherJob) // different job_id --> different token } - -func TestJobRunIdempotentCreate(t *testing.T) { - _, client := setupTestServerClient(t) - ctx := t.Context() - - // A run can only target an existing job, so create one first. - job, err := client.Jobs.Create(ctx, jobs.CreateJob{ - Name: "idempotency-job", - Tasks: []jobs.Task{{ - TaskKey: "t", - NotebookTask: &jobs.NotebookTask{NotebookPath: "/Workspace/Users/user@example.com/notebook"}, - }}, - }) - require.NoError(t, err) - - r := (&ResourceJobRun{}).New(client) - config := &JobRunState{RunNow: jobs.RunNow{JobId: job.JobId}} - - // Same config twice: the derived token is identical, so the backend returns - // the existing run instead of creating a duplicate. - id1, _, err := r.DoCreate(ctx, config) - require.NoError(t, err) - id2, _, err := r.DoCreate(ctx, config) - require.NoError(t, err) - assert.Equal(t, id1, id2, "same config must dedupe to the same run") - - // Different config: different token, so a genuinely new run is created. - other := &JobRunState{RunNow: jobs.RunNow{JobId: job.JobId, JobParameters: map[string]string{"env": "prod"}}} - id3, _, err := r.DoCreate(ctx, other) - require.NoError(t, err) - assert.NotEqual(t, id1, id3, "different config must create a new run") -} From 37568db8bae2ec9b32602f4d3c46d6b8950e7433 Mon Sep 17 00:00:00 2001 From: Rada Kamysheva Date: Mon, 13 Jul 2026 09:11:23 +0000 Subject: [PATCH 3/6] Regenerate out.test.toml for job_runs idempotent_recreate Pick up GOOSOnPR fields added by the acceptance harness on main (#5876, #5879), which restrict OS-independent bundle tests to Linux on PRs. --- .../bundle/resources/job_runs/idempotent_recreate/out.test.toml | 2 ++ 1 file changed, 2 insertions(+) diff --git a/acceptance/bundle/resources/job_runs/idempotent_recreate/out.test.toml b/acceptance/bundle/resources/job_runs/idempotent_recreate/out.test.toml index e90b6d5d1ba..1a3e24fa574 100644 --- a/acceptance/bundle/resources/job_runs/idempotent_recreate/out.test.toml +++ b/acceptance/bundle/resources/job_runs/idempotent_recreate/out.test.toml @@ -1,3 +1,5 @@ Local = true Cloud = false +GOOSOnPR.darwin = false +GOOSOnPR.windows = false EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] From 66ca15aa6a3dc8909a12da82a234fb33f7e136e6 Mon Sep 17 00:00:00 2001 From: Rada Kamysheva Date: Tue, 14 Jul 2026 08:29:23 +0000 Subject: [PATCH 4/6] Wait for job_runs completion and expose run output fields Add WaitAfterCreate to the job_runs direct-engine resource: it polls until the run reaches a terminal state and returns the run's output fields (state, run_id, run_page_url, ...) so downstream resources can read e.g. ${resources.job_runs.x.state.result_state}. A FAILED result is surfaced as readable state rather than failing the deploy; only INTERNAL_ERROR fails it. Add the wait_output acceptance test (a downstream job reads result_state, and a redeploy stays a no-op so the resolved value doesn't churn), and extend idempotent_recreate to cover both mid-run retry paths: a lost run id (token dedupes run-now) and a recorded run id (retry is a no-op). Regenerate the redeploy golden, which now shows the run TERMINATED. --- .../job_runs/idempotent_recreate/output.txt | 19 +++-- .../job_runs/idempotent_recreate/script | 27 +++--- .../resources/job_runs/redeploy/output.txt | 3 +- .../job_runs/wait_output/databricks.yml | 26 ++++++ .../job_runs/wait_output/out.test.toml | 5 ++ .../resources/job_runs/wait_output/output.txt | 84 +++++++++++++++++++ .../resources/job_runs/wait_output/script | 18 ++++ bundle/direct/dresources/job_run.go | 27 +++++- 8 files changed, 190 insertions(+), 19 deletions(-) create mode 100644 acceptance/bundle/resources/job_runs/wait_output/databricks.yml create mode 100644 acceptance/bundle/resources/job_runs/wait_output/out.test.toml create mode 100644 acceptance/bundle/resources/job_runs/wait_output/output.txt create mode 100644 acceptance/bundle/resources/job_runs/wait_output/script diff --git a/acceptance/bundle/resources/job_runs/idempotent_recreate/output.txt b/acceptance/bundle/resources/job_runs/idempotent_recreate/output.txt index d506c75a435..1cb7c2b4124 100644 --- a/acceptance/bundle/resources/job_runs/idempotent_recreate/output.txt +++ b/acceptance/bundle/resources/job_runs/idempotent_recreate/output.txt @@ -1,12 +1,12 @@ -=== initial deploy triggers the run +=== 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! -=== drop the job run from local state but keep the job (simulates a lost run id) +=== retry mid-run after a lost run id (deploy crashed before the id was recorded) >>> [CLI] bundle plan -o json { "depends_on": [ @@ -23,14 +23,23 @@ Deployment complete! } } -=== redeploy re-issues run-now; the token dedupes to the existing run >>> [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 twice with the same token, but no duplicate run was created +=== 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", @@ -48,7 +57,7 @@ Deployment complete! "job_id": [NUMID] } } -run id unchanged after recreate: the idempotency token deduped to the existing run +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: diff --git a/acceptance/bundle/resources/job_runs/idempotent_recreate/script b/acceptance/bundle/resources/job_runs/idempotent_recreate/script index e25372521fa..82750d55a08 100644 --- a/acceptance/bundle/resources/job_runs/idempotent_recreate/script +++ b/acceptance/bundle/resources/job_runs/idempotent_recreate/script @@ -9,25 +9,32 @@ run_id() { jq -r '.state["resources.job_runs.my_run"].__id__' "$STATE" } -title "initial deploy triggers the run" +title "initial deploy triggers the run and waits for it to finish" trace $CLI bundle deploy before=$(run_id) -title "drop the job run from local state but keep the job (simulates a lost run id)" -# The job stays deployed, so job_id -- and therefore the derived idempotency -# token -- is unchanged, which is what lets the retried run-now dedupe. +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 "redeploy re-issues run-now; the token dedupes to the existing run" +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=$(run_id) +after_kept=$(run_id) -title "run-now was issued twice with the same token, but no duplicate run was created" +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" ]; then - echo "run id unchanged after recreate: the idempotency token deduped to the existing run" +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): a duplicate run was created" + echo "run id changed ($before -> $after_lost -> $after_kept): a duplicate run was created" fi diff --git a/acceptance/bundle/resources/job_runs/redeploy/output.txt b/acceptance/bundle/resources/job_runs/redeploy/output.txt index b39382c0a4a..35339304c37 100644 --- a/acceptance/bundle/resources/job_runs/redeploy/output.txt +++ b/acceptance/bundle/resources/job_runs/redeploy/output.txt @@ -68,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": { 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..1a3e24fa574 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/wait_output/out.test.toml @@ -0,0 +1,5 @@ +Local = true +Cloud = false +GOOSOnPR.darwin = false +GOOSOnPR.windows = 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..f7cbfee9e59 --- /dev/null +++ b/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! 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..37bfa63428a --- /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 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 diff --git a/bundle/direct/dresources/job_run.go b/bundle/direct/dresources/job_run.go index d1ca0659153..c7a4bc8b8b8 100644 --- a/bundle/direct/dresources/job_run.go +++ b/bundle/direct/dresources/job_run.go @@ -7,6 +7,7 @@ import ( "encoding/json" "fmt" "strconv" + "time" "github.com/databricks/cli/bundle/config/resources" "github.com/databricks/databricks-sdk-go" @@ -14,6 +15,10 @@ import ( "github.com/databricks/databricks-sdk-go/service/jobs" ) +// jobRunTimeout bounds how long WaitAfterCreate waits for a run to finish. +// Matches the SDK's default RunNow waiter timeout. +const jobRunTimeout = 20 * time.Minute + // JobRunState is what we persist for a triggered run: the RunNow request. type JobRunState struct { jobs.RunNow @@ -150,12 +155,28 @@ func (r *ResourceJobRun) DoCreate(ctx context.Context, config *JobRunState) (str return strconv.FormatInt(wait.RunId, 10), nil, nil } +func (r *ResourceJobRun) WaitAfterCreate(ctx context.Context, id string, _ *JobRunState) (*JobRunRemote, error) { + runID, err := parseRunID(id) + if err != nil { + return nil, err + } + // TERMINATED/SKIPPED succeed even on a FAILED/TIMEDOUT/CANCELED result_state: + // the outcome is surfaced as readable state, not a deploy failure. Only + // INTERNAL_ERROR fails the deploy. + run, err := r.client.Jobs.WaitGetRunJobTerminatedOrSkipped(ctx, runID, jobRunTimeout, nil) + if err != nil { + return nil, err + } + return makeJobRunRemote(run), nil +} + // DoUpdate is intentionally not implemented: a run can't be modified in place, // so any change recreates it (delete + a fresh RunNow). -// 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. +// DoDelete deletes the run via jobs/runs/delete, on both destroy and recreate. +// WaitAfterCreate leaves a run terminal by the end of its deploy, so on recreate +// the prior run is safe to delete; destroying a run still active from an +// interrupted deploy is rejected by the API and surfaces to the user. func (r *ResourceJobRun) DoDelete(ctx context.Context, id string, _ *JobRunState) error { runID, err := parseRunID(id) if err != nil { From 3d743393c054e9927f1209634200e5dac63f704e Mon Sep 17 00:00:00 2001 From: Rada Kamysheva Date: Tue, 14 Jul 2026 10:08:54 +0000 Subject: [PATCH 5/6] Add job_runs changelog and fix WaitAfterCreate test assertion Add a changelog fragment for the idempotency + wait-for-completion work. The shared CRUD harness compared WaitAfterCreate's result against the read taken right after DoCreate. A job run transitions RUNNING -> TERMINATED while we wait, so compare against a fresh read taken after the wait instead, mirroring how DoUpdate is already validated. --- .nextchanges/bundles/job-runs-idempotency-wait.md | 1 + bundle/direct/dresources/all_test.go | 8 +++++++- 2 files changed, 8 insertions(+), 1 deletion(-) create mode 100644 .nextchanges/bundles/job-runs-idempotency-wait.md diff --git a/.nextchanges/bundles/job-runs-idempotency-wait.md b/.nextchanges/bundles/job-runs-idempotency-wait.md new file mode 100644 index 00000000000..ba8d8398df8 --- /dev/null +++ b/.nextchanges/bundles/job-runs-idempotency-wait.md @@ -0,0 +1 @@ +direct: the experimental `job_runs` resource now dedupes runs across deploy retries with an automatically computed `idempotency_token`, waits for the run to reach a terminal state, and exposes its output fields (e.g. `${resources.job_runs..state.result_state}`) for use by downstream resources. diff --git a/bundle/direct/dresources/all_test.go b/bundle/direct/dresources/all_test.go index d79c80c61b2..d4d3e9b9ea1 100644 --- a/bundle/direct/dresources/all_test.go +++ b/bundle/direct/dresources/all_test.go @@ -985,7 +985,13 @@ func testCRUD(t *testing.T, group string, adapter *Adapter, client *databricks.W remoteStateFromWaitCreate, err := adapter.WaitAfterCreate(ctx, createdID, newState) require.NoError(t, err) if remoteStateFromWaitCreate != nil { - require.Equal(t, remote, remoteStateFromWaitCreate) + // WaitAfterCreate returns the settled remote, which may differ from the + // read taken right after DoCreate (a job run transitions from RUNNING to + // TERMINATED while we wait). Compare against a fresh read taken after the + // wait, which reflects that same settled state. + remotePostWaitCreate, err := adapter.DoRead(ctx, createdID) + require.NoError(t, err) + require.Equal(t, remotePostWaitCreate, remoteStateFromWaitCreate) } if adapter.HasDoUpdate() { From 7faf08b90491bc29cf299437418c99df29e95bed Mon Sep 17 00:00:00 2001 From: Rada Kamysheva Date: Tue, 14 Jul 2026 10:59:21 +0000 Subject: [PATCH 6/6] Shorten job_runs changelog entry --- .nextchanges/bundles/job-runs-idempotency-wait.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.nextchanges/bundles/job-runs-idempotency-wait.md b/.nextchanges/bundles/job-runs-idempotency-wait.md index ba8d8398df8..b1ca338e76e 100644 --- a/.nextchanges/bundles/job-runs-idempotency-wait.md +++ b/.nextchanges/bundles/job-runs-idempotency-wait.md @@ -1 +1 @@ -direct: the experimental `job_runs` resource now dedupes runs across deploy retries with an automatically computed `idempotency_token`, waits for the run to reach a terminal state, and exposes its output fields (e.g. `${resources.job_runs..state.result_state}`) for use by downstream resources. +direct: the experimental `job_runs` resource now waits for the run to finish, exposes its output fields (e.g. `${resources.job_runs..state.result_state}`), and dedupes runs on deploy retries.