bundle: extract DeploymentLock interface + workspace filesystem impl

bundle/bundle.go

@@ -21,7 +21,6 @@ import (
 	"github.com/databricks/cli/libs/auth"
 	"github.com/databricks/cli/libs/cache"
 	"github.com/databricks/cli/libs/fileset"
-	"github.com/databricks/cli/libs/locker"
 	"github.com/databricks/cli/libs/log"
 	"github.com/databricks/cli/libs/logdiag"
 	libsync "github.com/databricks/cli/libs/sync"
@@ -129,9 +128,6 @@ type Bundle struct {
 	// Stores an initialized copy of this bundle's Terraform wrapper.
 	Terraform *tfexec.Terraform
 
-	// Stores the locker responsible for acquiring/releasing a deployment lock.
-	Locker *locker.Locker
-
 	// TerraformPlanPath is the path to the plan from the terraform CLI
 	TerraformPlanPath string
 

bundle/deploy/lock/lock.go

@@ -0,0 +1,59 @@
+package lock
+
+import (
+	"context"
+
+	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/bundle/permissions"
+	"github.com/databricks/cli/libs/diag"
+)
+
+// Goal describes the purpose of a deployment operation.
+type Goal string
+
+const (
+	GoalBind    = Goal("bind")
+	GoalUnbind  = Goal("unbind")
+	GoalDeploy  = Goal("deploy")
+	GoalDestroy = Goal("destroy")
+)
+
+// DeploymentStatus indicates whether the deployment operation succeeded or failed.
+type DeploymentStatus int
+
+const (
+	DeploymentSuccess DeploymentStatus = iota
+	DeploymentFailure
+)
+
+// DeploymentLock manages the deployment lock lifecycle.
+type DeploymentLock interface {
+	// Acquire acquires the deployment lock.
+	Acquire(ctx context.Context) error
+
+	// Release releases the deployment lock with the given deployment status.
+	Release(ctx context.Context, status DeploymentStatus) error
+}
+
+// NewDeploymentLock returns a DeploymentLock backed by the workspace
+// filesystem. Captures everything the lock needs from the bundle at
+// construction time so the lock implementation itself does not retain a
+// *bundle.Bundle reference. The workspace client is only initialized when
+// locking is enabled to match the original lazy-init behavior.
+func NewDeploymentLock(ctx context.Context, b *bundle.Bundle, goal Goal) DeploymentLock {
+	enabled := b.Config.Bundle.Deployment.Lock.IsEnabled()
+	l := &workspaceFilesystemLock{
+		user:      b.Config.Workspace.CurrentUser.UserName,
+		statePath: b.Config.Workspace.StatePath,
+		enabled:   enabled,
+		force:     b.Config.Bundle.Deployment.Lock.Force,
+		goal:      goal,
+		reportPermissionError: func(ctx context.Context, path string) diag.Diagnostics {
+			return permissions.ReportPossiblePermissionDenied(ctx, b, path)
+		},
+	}
+	if enabled {
+		l.client = b.WorkspaceClient(ctx)
+	}
+	return l
+}

bundle/deploy/lock/workspace_filesystem.go

@@ -0,0 +1,83 @@
+package lock
+
+import (
+	"context"
+	"errors"
+	"io/fs"
+
+	"github.com/databricks/cli/libs/diag"
+	"github.com/databricks/cli/libs/locker"
+	"github.com/databricks/cli/libs/log"
+	"github.com/databricks/databricks-sdk-go"
+)
+
+// workspaceFilesystemLock implements DeploymentLock using a lock file in the
+// bundle's workspace state path. Holds only the primitives it needs from the
+// bundle.
+type workspaceFilesystemLock struct {
+	client    *databricks.WorkspaceClient
+	user      string
+	statePath string
+	enabled   bool
+	force     bool
+
+	// reportPermissionError produces the user-facing permission diagnostic
+	// when the workspace API returns ErrPermission/ErrNotExist from Lock.
+	// Lifted to a callback so this struct does not pin a *bundle.Bundle.
+	reportPermissionError func(ctx context.Context, path string) diag.Diagnostics
+
+	locker *locker.Locker
+	goal   Goal
+}
+
+func (l *workspaceFilesystemLock) Acquire(ctx context.Context) error {
+	// Return early if locking is disabled.
+	if !l.enabled {
+		log.Infof(ctx, "Skipping; locking is disabled")
+		return nil
+	}
+
+	lk, err := locker.CreateLocker(l.user, l.statePath, l.client)
+	if err != nil {
+		return err
+	}
+
+	l.locker = lk
+
+	log.Infof(ctx, "Acquiring deployment lock (force: %v)", l.force)
+	err = lk.Lock(ctx, l.force)
+	if err != nil {
+		log.Errorf(ctx, "Failed to acquire deployment lock: %v", err)
+
+		// If we get a permission or "doesn't exist" error from the API this
+		// indicates we either don't have permissions or the path is invalid.
+		if errors.Is(err, fs.ErrPermission) || errors.Is(err, fs.ErrNotExist) {
+			return l.reportPermissionError(ctx, l.statePath).Error()
+		}
+
+		return err
+	}
+
+	return nil
+}
+
+func (l *workspaceFilesystemLock) Release(ctx context.Context, _ DeploymentStatus) error {
+	// Return early if locking is disabled.
+	if !l.enabled {
+		log.Infof(ctx, "Skipping; locking is disabled")
+		return nil
+	}
+
+	// Return early if the locker is not set.
+	// It is likely an error occurred prior to initialization of the locker instance.
+	if l.locker == nil {
+		log.Warnf(ctx, "Unable to release lock if locker is not configured")
+		return nil
+	}
+
+	log.Infof(ctx, "Releasing deployment lock")
+	if l.goal == GoalDestroy {
+		return l.locker.Unlock(ctx, locker.AllowLockFileNotExist)
+	}
+	return l.locker.Unlock(ctx)
+}

bundle/phases/bind.go

@@ -23,13 +23,20 @@ import (
 func Bind(ctx context.Context, b *bundle.Bundle, opts *terraform.BindOptions, engine engine.EngineType) {
 	log.Info(ctx, "Phase: bind")
 
-	bundle.ApplyContext(ctx, b, lock.Acquire())
-	if logdiag.HasError(ctx) {
+	dl := lock.NewDeploymentLock(ctx, b, lock.GoalBind)
+	if err := dl.Acquire(ctx); err != nil {
+		logdiag.LogError(ctx, err)
 		return
 	}
 
 	defer func() {
-		bundle.ApplyContext(ctx, b, lock.Release(lock.GoalBind))
+		status := lock.DeploymentSuccess
+		if logdiag.HasError(ctx) {
+			status = lock.DeploymentFailure
+		}
+		if err := dl.Release(ctx, status); err != nil {
+			logdiag.LogError(ctx, err)
+		}
 	}()
 
 	if engine.IsDirect() {
@@ -119,13 +126,20 @@ func jsonDump(ctx context.Context, v any, field string) string {
 func Unbind(ctx context.Context, b *bundle.Bundle, bundleType, tfResourceType, resourceKey string, engine engine.EngineType) {
 	log.Info(ctx, "Phase: unbind")
 
-	bundle.ApplyContext(ctx, b, lock.Acquire())
-	if logdiag.HasError(ctx) {
+	dl := lock.NewDeploymentLock(ctx, b, lock.GoalUnbind)
+	if err := dl.Acquire(ctx); err != nil {
+		logdiag.LogError(ctx, err)
 		return
 	}
 
 	defer func() {
-		bundle.ApplyContext(ctx, b, lock.Release(lock.GoalUnbind))
+		status := lock.DeploymentSuccess
+		if logdiag.HasError(ctx) {
+			status = lock.DeploymentFailure
+		}
+		if err := dl.Release(ctx, status); err != nil {
+			logdiag.LogError(ctx, err)
+		}
 	}()
 
 	if engine.IsDirect() {

bundle/phases/deploy.go

@@ -125,19 +125,27 @@ func Deploy(ctx context.Context, b *bundle.Bundle, outputHandler sync.OutputHand
 
 	// Core mutators that CRUD resources and modify deployment state. These
 	// mutators need informed consent if they are potentially destructive.
-	bundle.ApplySeqContext(ctx, b,
-		scripts.Execute(config.ScriptPreDeploy),
-		lock.Acquire(),
-	)
-
+	bundle.ApplyContext(ctx, b, scripts.Execute(config.ScriptPreDeploy))
 	if logdiag.HasError(ctx) {
 		// lock is not acquired here
 		return
 	}
 
+	dl := lock.NewDeploymentLock(ctx, b, lock.GoalDeploy)
+	if err := dl.Acquire(ctx); err != nil {
+		logdiag.LogError(ctx, err)
+		return
+	}
+
 	// lock is acquired here
 	defer func() {
-		bundle.ApplyContext(ctx, b, lock.Release(lock.GoalDeploy))
+		status := lock.DeploymentSuccess
+		if logdiag.HasError(ctx) {
+			status = lock.DeploymentFailure
+		}
+		if err := dl.Release(ctx, status); err != nil {
+			logdiag.LogError(ctx, err)
+		}
 	}()
 
 	uploadLibraries(ctx, b, libs)

bundle/phases/destroy.go

@@ -120,13 +120,20 @@ func Destroy(ctx context.Context, b *bundle.Bundle, engine engine.EngineType) {
 		return
 	}
 
-	bundle.ApplyContext(ctx, b, lock.Acquire())
-	if logdiag.HasError(ctx) {
+	dl := lock.NewDeploymentLock(ctx, b, lock.GoalDestroy)
+	if err := dl.Acquire(ctx); err != nil {
+		logdiag.LogError(ctx, err)
 		return
 	}
 
 	defer func() {
-		bundle.ApplyContext(ctx, b, lock.Release(lock.GoalDestroy))
+		status := lock.DeploymentSuccess
+		if logdiag.HasError(ctx) {
+			status = lock.DeploymentFailure
+		}
+		if err := dl.Release(ctx, status); err != nil {
+			logdiag.LogError(ctx, err)
+		}
 	}()
 
 	if !engine.IsDirect() {