direct: add DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT to cap resource waits

.nextchanges/bundles/resource-max-wait.md

@@ -0,0 +1 @@
+Added `DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT` (in seconds) to cap how long deploy and destroy wait for a resource. Direct engine only.

acceptance/bundle/deploy/resource-max-wait-zero/databricks.yml

@@ -0,0 +1,15 @@
+bundle:
+  name: test-bundle
+
+# Nothing to sync: keeps the "Files:" line out of the output, which would otherwise
+# count whatever the harness has written into the test directory.
+sync:
+  paths: []
+
+resources:
+  clusters:
+    my_cluster:
+      cluster_name: test-cluster
+      spark_version: 13.3.x-scala2.12
+      node_type_id: i3.xlarge
+      num_workers: 1

acceptance/bundle/deploy/resource-max-wait-zero/output.txt

@@ -0,0 +1,38 @@
+
+=== Zero cap: the create wait is skipped entirely
+>>> DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT=0 [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files...
+Warn: deploying resources.clusters.my_cluster: Not waiting for creation of resources.clusters.my_cluster (DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT=0); it may still be in progress
+Created clusters.my_cluster
+Files: 0 uploaded, 0 deleted
+Resources: 1 created, 0 changed, 0 deleted, 0 unchanged
+
+>>> print_requests.py --get --unique --oneline --del-field body,q //clusters
+{"method": "POST", "path": "/api/2.1/clusters/create"}
+
+>>> [CLI] bundle destroy --auto-approve
+The following resources will be deleted:
+  delete resources.clusters.my_cluster
+
+All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/test-bundle/default
+
+Destroy: 1 deleted
+
+=== No cap: the deployment polls until the cluster is RUNNING
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files...
+Created clusters.my_cluster
+Files: 0 uploaded, 0 deleted
+Resources: 1 created, 0 changed, 0 deleted, 0 unchanged
+
+>>> print_requests.py --get --unique --oneline --del-field body,q //clusters
+{"method": "POST", "path": "/api/2.1/clusters/create"}
+{"method": "GET", "path": "/api/2.1/clusters/get"}
+
+>>> [CLI] bundle destroy --auto-approve
+The following resources will be deleted:
+  delete resources.clusters.my_cluster
+
+All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/test-bundle/default
+
+Destroy: 1 deleted

acceptance/bundle/deploy/resource-max-wait-zero/script

@@ -0,0 +1,23 @@
+cleanup() {
+    rm -f out.requests.txt
+}
+trap cleanup EXIT
+
+title "Zero cap: the create wait is skipped entirely"
+rm -f out.requests.txt
+trace DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT=0 $CLI bundle deploy
+
+# Only the create POST is recorded. No GET means the cluster was left PENDING: the
+# testserver advances PENDING -> RUNNING on read, so a wait would have shown up here.
+# Method and path are the whole assertion; the body would only add churn.
+trace print_requests.py --get --unique --oneline --del-field body,q //clusters
+
+trace $CLI bundle destroy --auto-approve
+
+title "No cap: the deployment polls until the cluster is RUNNING"
+rm -f out.requests.txt
+trace $CLI bundle deploy
+
+trace print_requests.py --get --unique --oneline --del-field body,q //clusters
+
+trace $CLI bundle destroy --auto-approve

acceptance/bundle/deploy/resource-max-wait-zero/test.toml

@@ -0,0 +1,9 @@
+# The cap is read by the direct engine only.
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
+
+RecordRequests = true
+
+# A cluster is created PENDING and only reaches RUNNING when something polls it
+# (see ClustersGet in libs/testserver), so the poll GET is what proves whether the
+# deployment waited.
+Ignore = [".databricks"]

acceptance/bundle/deploy/resource-max-wait/databricks.yml

@@ -0,0 +1,16 @@
+bundle:
+  name: test-bundle
+
+# Nothing to sync: keeps the "Files:" line out of the output, which would otherwise
+# count whatever the harness has written into the test directory.
+sync:
+  paths: []
+
+resources:
+  jobs:
+    my_job:
+      name: test-job
+      tasks:
+        - task_key: main
+          notebook_task:
+            notebook_path: /Workspace/notebook

acceptance/bundle/deploy/resource-max-wait/output.txt

@@ -0,0 +1,28 @@
+
+=== A malformed value is rejected instead of falling back to the default wait
+>>> DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT=1m musterr [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files...
+Error: invalid DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT="1m": expected a non-negative number of seconds
+
+Files: 0 uploaded, 0 deleted
+
+>>> DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT=-5 musterr [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files...
+Error: invalid DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT="-5": expected a non-negative number of seconds
+
+Files: 0 uploaded, 0 deleted
+
+=== A valid value is accepted
+>>> DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT=60 [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files...
+Created jobs.my_job
+Files: 0 uploaded, 0 deleted
+Resources: 1 created, 0 changed, 0 deleted, 0 unchanged
+
+>>> [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/test-bundle/default
+
+Destroy: 1 deleted

acceptance/bundle/deploy/resource-max-wait/script

@@ -0,0 +1,8 @@
+title "A malformed value is rejected instead of falling back to the default wait"
+trace DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT=1m musterr $CLI bundle deploy
+trace DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT=-5 musterr $CLI bundle deploy
+
+title "A valid value is accepted"
+trace DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT=60 $CLI bundle deploy
+
+trace $CLI bundle destroy --auto-approve

acceptance/bundle/deploy/resource-max-wait/test.toml

@@ -0,0 +1,5 @@
+# Covers parsing of DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT only. What the cap does to a wait
+# is covered by ../resource-max-wait-zero, whose resource actually has one: jobs have no
+# WaitAfterCreate, so a cap would be a no-op here.
+# DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT is read by the direct engine only.
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]

acceptance/bundle/invariant/configs/vector_search_index.yml.tmpl-init.sh

@@ -0,0 +1,12 @@
+#!/bin/bash
+
+# Index provisioning takes 15-30 minutes, which is why this config used to be excluded from
+# cloud runs entirely. None of the invariants need a queryable index -- they deploy, re-plan,
+# delete and re-delete -- so cap the wait instead of skipping the coverage.
+#
+# Sourced by invariant_render, so the export applies to every $CLI call in the script.
+#
+# Deliberately per-config rather than Env in the directory's test.toml: this index is three
+# orders of magnitude slower than the next slowest config (cluster, ~100s), and capping every
+# config would leave nothing exercising a real WaitAfterCreate against a real backend.
+export DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT=30

acceptance/bundle/invariant/test.toml

@@ -104,11 +104,6 @@ EnvMatrixExclude.no_catalog_optional_fields_on_cloud = ["CONFIG_Cloud=true", "IN
 EnvMatrixExclude.no_external_location_on_cloud = ["CONFIG_Cloud=true", "INPUT_CONFIG=external_location.yml.tmpl"]
 # External volumes reference external locations; excluded from cloud for the same reason
 EnvMatrixExclude.no_external_volume_on_cloud = ["CONFIG_Cloud=true", "INPUT_CONFIG=volume_external.yml.tmpl"]
-# A vector search index takes 10-20 min to create, and these tests deploy it twice
-# (deploy, then re-plan), making it by far the slowest variant on cloud. The dedicated
-# vector_search_indexes resource test already covers it on cloud (CloudSlow), so exclude
-# it here to keep the cloud invariant runs from timing out. Still exercised locally.
-EnvMatrixExclude.no_vector_search_index_on_cloud = ["CONFIG_Cloud=true", "INPUT_CONFIG=vector_search_index.yml.tmpl"]
 # Telemetry requires a real model, which cloud invariant tests do not provision.
 EnvMatrixExclude.no_model_serving_endpoint_telemetry_on_cloud = ["CONFIG_Cloud=true", "INPUT_CONFIG=model_serving_endpoint_telemetry.yml.tmpl"]
 # Fake SQL endpoint for local tests

acceptance/bundle/resources/vector_search_indexes/basic/test.toml

@@ -0,0 +1,17 @@
+# Index provisioning takes ~30 minutes, which exceeded the test timeout and made this the
+# slowest test in the cloud job. Nothing here depends on the index being queryable — the
+# assertions only need it to exist — so cap the wait instead. The cap is what this test's
+# runtime now consists of (~2x cap: once on create, once on delete), so keep it small.
+Env.DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT = "30"
+
+# With the wait capped this is no longer slow, so run it on every PR rather than only in the
+# nightly job. The parent sets CloudSlow=true for the tests that still wait ~30 minutes.
+CloudSlow = false
+Cloud = true
+
+# The cap makes deploy warn that it stopped waiting. Drop the line rather than baking it into
+# the golden: it is a property of this test's configuration, not of the behaviour under test.
+# Trailing \n so the whole line goes, leaving no blank line behind.
+[[Repls]]
+Old = 'Warn: [^\n]*Stopped waiting[^\n]*\n'
+New = ''

acceptance/bundle/resources/vector_search_indexes/grants/select/test.toml

@@ -1,2 +1,15 @@
 RequiresUnityCatalog = true
 RecordRequests = true
+
+# The index's only dependent is its own grants child, which needs the index id but not a
+# provisioned index, so the wait is capped here too. Runtime becomes ~2x the cap.
+Env.DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT = "30"
+
+# No longer slow once the wait is capped, so run it on every PR.
+CloudSlow = false
+Cloud = true
+
+# See basic/test.toml: the cap makes deploy warn that it stopped waiting.
+[[Repls]]
+Old = 'Warn: [^\n]*Stopped waiting[^\n]*\n'
+New = ''

bundle/direct/apply.go

@@ -88,8 +88,10 @@ func (d *DeploymentUnit) Create(ctx context.Context, db *dstate.DeploymentState,
 		return fmt.Errorf("saving state after creating id=%s: %w", newID, err)
 	}
 
-	waitRemoteState, err := retryOnTransient(ctx, func() (any, error) {
-		return d.Adapter.WaitAfterCreate(ctx, newID, newState)
+	waitRemoteState, err := waitCapped(ctx, d.MaxWait, "creation of "+d.ResourceKey, func(ctx context.Context) (any, error) {
+		return retryOnTransient(ctx, func() (any, error) {
+			return d.Adapter.WaitAfterCreate(ctx, newID, newState)
+		})
 	})
 	if err != nil {
 		return fmt.Errorf("waiting after creating id=%s: %w", newID, err)
@@ -266,7 +268,11 @@ func (d *DeploymentUnit) Delete(ctx context.Context, db *dstate.DeploymentState,
 	// Wait for asynchronous teardown after dropping state. Mirrors Recreate so
 	// the contract is the same regardless of whether the user triggered
 	// `bundle destroy` or a recreate.
-	err = d.Adapter.WaitAfterDelete(ctx, oldID)
+	// The two diverge once MaxWait is set: this wait is capped, Recreate's is not,
+	// because only Recreate needs the name released for the create that follows.
+	_, err = waitCapped(ctx, d.MaxWait, "deletion of "+d.ResourceKey, func(ctx context.Context) (struct{}, error) {
+		return struct{}{}, d.Adapter.WaitAfterDelete(ctx, oldID)
+	})
 	if err != nil {
 		return fmt.Errorf("waiting after deleting id=%s: %w", oldID, err)
 	}

bundle/direct/bundle_apply.go

@@ -9,6 +9,7 @@ import (
 	"github.com/databricks/cli/bundle/config"
 	"github.com/databricks/cli/bundle/deployplan"
 	"github.com/databricks/cli/bundle/terraform_dabs_map"
+	"github.com/databricks/cli/libs/log"
 	"github.com/databricks/cli/libs/logdiag"
 	"github.com/databricks/cli/libs/structs/structaccess"
 	"github.com/databricks/cli/libs/structs/structpath"
@@ -20,6 +21,14 @@ func (b *DeploymentBundle) Apply(ctx context.Context, client *databricks.Workspa
 		panic("Planning is not done")
 	}
 
+	// Read before the early return below so a malformed value is reported even when there is
+	// nothing to deploy.
+	maxWait, err := resourceMaxWait(ctx)
+	if err != nil {
+		logdiag.LogError(ctx, err)
+		return
+	}
+
 	if len(plan.Plan) == 0 {
 		// Avoid creating state file if nothing to deploy
 		return
@@ -70,10 +79,24 @@ func (b *DeploymentBundle) Apply(ctx context.Context, client *databricks.Workspa
 			return false
 		}
 
+		// Deletes are capped even with dependents: state is dropped before the wait, so a
+		// cut-short delete leaves the resource untracked while it tears down, and a dependency
+		// deleted after it may be rejected for still having a child. Accepted deliberately.
+		// Recreate's internal delete-wait is never routed through the cap at all, because it
+		// releases the name for the create that follows.
+		unitWait := maxWait
+		if action != deployplan.Delete && hasBlockingDependents(g, resourceKey) {
+			unitWait = maxWaitUnset
+			if maxWait != maxWaitUnset {
+				log.Debugf(ctx, "Not capping wait for %s: other resources depend on it", resourceKey)
+			}
+		}
+
 		d := &DeploymentUnit{
 			ResourceKey: resourceKey,
 			Adapter:     adapter,
 			DependsOn:   entry.DependsOn,
+			MaxWait:     unitWait,
 		}
 
 		if action == deployplan.Delete {

bundle/direct/maxwait.go

@@ -0,0 +1,96 @@
+package direct
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strconv"
+	"strings"
+	"time"
+
+	bundleenv "github.com/databricks/cli/bundle/env"
+	"github.com/databricks/cli/libs/dagrun"
+	"github.com/databricks/cli/libs/log"
+	"github.com/databricks/databricks-sdk-go/retries"
+)
+
+// maxWaitUnset means "no cap configured", which must stay distinguishable from an explicit
+// 0 ("do not wait at all").
+const maxWaitUnset = time.Duration(-1)
+
+// resourceMaxWait returns the cap on waiting for a resource to reach its target state, or
+// maxWaitUnset when the environment variable is absent. Unlike retryInterval, a malformed
+// value is an error rather than a silent fallback: ignoring a typo would restore the
+// multi-hour default wait that the user was trying to shorten.
+func resourceMaxWait(ctx context.Context) (time.Duration, error) {
+	v, ok := bundleenv.ResourceMaxWait(ctx)
+	if !ok {
+		return maxWaitUnset, nil
+	}
+	seconds, err := strconv.Atoi(v)
+	if err != nil || seconds < 0 {
+		return maxWaitUnset, fmt.Errorf("invalid %s=%q: expected a non-negative number of seconds", bundleenv.ResourceMaxWaitVariable, v)
+	}
+	return time.Duration(seconds) * time.Second, nil
+}
+
+// hasBlockingDependents reports whether any node that runs after resourceKey needs it to have
+// reached its target state.
+//
+// Child nodes (.permissions, .grants) are excluded: they reference nothing but the parent's id
+// (see PrepareGrantsInputConfig and PreparePermissionsInputConfig), which DoCreate returns
+// before the wait even starts, so they attach to a resource that exists but is not yet
+// provisioned. Only a 4-segment key can have a 3-segment resource key as its prefix, so the
+// prefix test cannot match a sibling.
+func hasBlockingDependents(g *dagrun.Graph, resourceKey string) bool {
+	for _, edge := range g.Adj[resourceKey] {
+		if !strings.HasPrefix(edge.To, resourceKey+".") {
+			return true
+		}
+	}
+	return false
+}
+
+// waitCapped runs wait under maxWait. When the cap expires the wait is abandoned with a
+// warning instead of failing the deployment: state is written before the wait, so the
+// resource stays tracked and the next plan reconciles it. Genuine failures still propagate,
+// since retries reports those without a timeout error.
+func waitCapped[T any](ctx context.Context, maxWait time.Duration, description string, wait func(context.Context) (T, error)) (T, error) {
+	if maxWait == maxWaitUnset {
+		return wait(ctx)
+	}
+
+	if maxWait == 0 {
+		// Skip the call rather than starting a poll that is already out of time, which would
+		// spend one request to learn what the caller has already said it does not care about.
+		log.Warnf(ctx, "Not waiting for %s (%s=0); it may still be in progress", description, bundleenv.ResourceMaxWaitVariable)
+		var zero T
+		return zero, nil
+	}
+
+	waitCtx, cancel := context.WithTimeout(ctx, maxWait)
+	defer cancel()
+
+	result, err := wait(waitCtx)
+
+	// waitCtx expired but ctx did not: the cap fired rather than the whole deployment being
+	// cancelled, which must keep failing so an interrupt is not swallowed.
+	if err != nil && waitCtx.Err() != nil && ctx.Err() == nil && isWaitTimeout(err) {
+		log.Warnf(ctx, "Stopped waiting for %s after %s (%s); it may still be in progress", description, maxWait, bundleenv.ResourceMaxWaitVariable)
+		var zero T
+		return zero, nil
+	}
+
+	return result, err
+}
+
+// isWaitTimeout reports whether err is a wait that ran out of time rather than a resource
+// that failed. Two shapes reach here: retries.Poll reports a deadline as ErrTimedOut wrapping
+// the last poll message, while retryWith returns a bare context error when the deadline lands
+// while it sleeps between transient-error retries.
+func isWaitTimeout(err error) bool {
+	if _, ok := errors.AsType[*retries.ErrTimedOut](err); ok {
+		return true
+	}
+	return errors.Is(err, context.DeadlineExceeded)
+}