bundle/deploy/lock: introduce DeploymentManager interface aligned with DMS versioning model

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,68 @@
+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 lifecycle of a bundle deployment.
+// The workspace-filesystem lock serializes concurrent deployments.
+// DMS version tracking will be added additively — see deployment_metadata_service.go.
+type DeploymentLock struct {
+	wfs workspaceFilesystemLock
+}
+
+// NewDeploymentLock returns a DeploymentLock for the bundle.
+// Captures everything it needs from the bundle at construction time
+// so the lock does not retain a *bundle.Bundle reference. The
+// workspace client is only initialized when locking is enabled.
+func NewDeploymentLock(ctx context.Context, b *bundle.Bundle, goal Goal) *DeploymentLock {
+	enabled := b.Config.Bundle.Deployment.Lock.IsEnabled()
+	l := &DeploymentLock{
+		wfs: 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.wfs.client = b.WorkspaceClient(ctx)
+	}
+	return l
+}
+
+// Acquire acquires the deployment lock.
+func (l *DeploymentLock) Acquire(ctx context.Context) error {
+	return l.wfs.acquire(ctx)
+}
+
+// Release releases the deployment lock.
+func (l *DeploymentLock) Release(ctx context.Context, status DeploymentStatus) error {
+	return l.wfs.release(ctx, status)
+}

bundle/deploy/lock/workspace_filesystem.go

@@ -0,0 +1,84 @@
+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 holds the state for the workspace-filesystem lock.
+// Methods are unexported; callers use DeploymentLock.Acquire / DeploymentLock.Release.
+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 {
+		// AllowLockFileNotExist because the destroy phase deletes the remote
+		// state directory, which includes the lock file itself.
+		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

@@ -126,19 +126,26 @@ 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
+		// lock not yet acquired
+		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)