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/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/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..8c10d8df85f --- /dev/null +++ b/bundle/direct/dresources/job_run_test.go @@ -0,0 +1,46 @@ +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 +} 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 c3537d00187..43a3342845f 100644 --- a/libs/testserver/jobs.go +++ b/libs/testserver/jobs.go @@ -348,6 +348,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 != "" { @@ -411,6 +419,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}} }