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 `bundle deploy` and `bundle destroy` wait for a resource to become ready or finish deleting. When the cap expires the wait is abandoned with a warning instead of failing — the resource is already recorded in state, so the next deployment reconciles it. Resources that others depend on keep their full wait, since a dependent must not act on a resource that is not ready. Direct engine only.

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

@@ -0,0 +1,11 @@
+bundle:
+  name: test-bundle
+
+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,36 @@
+
+=== 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...
+Deploying resources...
+Error: invalid DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT="1m": expected a non-negative number of seconds
+
+
+>>> DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT=-5 musterr [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files...
+Deploying resources...
+Error: invalid DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT="-5": expected a non-negative number of seconds
+
+
+=== A valid value deploys normally
+>>> DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT=60 [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!
+
+=== Zero means do not wait at all
+>>> DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT=0 [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!
+
+>>> [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
+
+Deleting files...
+Destroy complete!

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

@@ -0,0 +1,11 @@
+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 deploys normally"
+trace DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT=60 $CLI bundle deploy
+
+title "Zero means do not wait at all"
+trace DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT=0 $CLI bundle deploy
+
+trace $CLI bundle destroy --auto-approve

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

@@ -0,0 +1,2 @@
+# DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT is read by the direct engine only.
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]

acceptance/bundle/resources/vector_search_indexes/test.toml

@@ -1,8 +1,11 @@
 Cloud = true
 RequiresUnityCatalog = true
 
+# EXPERIMENT (do not merge): CloudSlow=false so these run in the short PR integration job,
+# to measure how DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT changes their wall clock.
+Env.DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT = "300"
 # Vector Search indexes are slow to create, so only run them in the nightly (non-short) cloud job.
-CloudSlow = true
+CloudSlow = false
 
 # Vector Search indexes are only available in direct mode (no Terraform provider)
 EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]

bundle/direct/apply.go

@@ -80,8 +80,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)
@@ -258,7 +260,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,17 @@ func (b *DeploymentBundle) Apply(ctx context.Context, client *databricks.Workspa
 			return false
 		}
 
+		// g.Adj holds the edges out of this node, i.e. the resources that run after it.
+		unitWait := unitMaxWait(maxWait, action, len(g.Adj[resourceKey]))
+		if maxWait != maxWaitUnset && unitWait == 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,88 @@
+package direct
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strconv"
+	"time"
+
+	"github.com/databricks/cli/bundle/deployplan"
+	bundleenv "github.com/databricks/cli/bundle/env"
+	"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
+}
+
+// unitMaxWait returns the cap to apply to a single resource, given how many resources run
+// after it in the deployment graph.
+//
+// Deletes are capped regardless of dependents. The trade-off is accepted deliberately: state
+// is dropped before the wait, so a cut-short delete leaves the resource untracked while it is
+// still tearing down, and the dependency deleted after it may then be rejected for still
+// having a child. Recreate's internal delete-wait is excluded structurally — it is the one
+// wait never routed through here, because it releases the name for the following create.
+//
+// Every other action is capped only when nothing depends on this resource, since a dependent
+// would otherwise act on a resource that has not reached its target state.
+func unitMaxWait(maxWait time.Duration, action deployplan.ActionType, dependents int) time.Duration {
+	if action == deployplan.Delete || dependents == 0 {
+		return maxWait
+	}
+	return maxWaitUnset
+}
+
+// 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)
+	}
+
+	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)
+}