bundle/deploy/lock: record deployment history in DMS behind experimental.record_deployment_history

Adds a deploymentVersionRecorder that records each deploy/destroy as a version
with the Deployment Metadata Service (DMS), running alongside the existing
workspace-filesystem lock inside DeploymentLock.Acquire / Release. Enabled by
the experimental.record_deployment_history config field.

The deployment ID is the state lineage read from resources.json, so a bundle
deployment maps one-to-one to a DMS deployment record:

- Acquire: after the file lock, GetDeployment(lineage); CreateDeployment only
  when it does not exist yet; then create the next version (server validates
  version_id == last_version_id + 1). A heartbeat goroutine keeps the version's
  lease alive. dstate.GetOrInitLineage generates and stores a lineage on the
  first deploy so the deploy persists the same value.
- Release: complete the version (success/failure); on destroy + success, delete
  the deployment record.

bind/unbind and --force are not supported with DMS. The version recorder uses
w.Bundle from the Databricks Go SDK directly. Adds the config field, its schema
annotation, and regenerates jsonschema.json.

Co-authored-by: Shreyas Goenka <shreyas.goenka@databricks.com>

Author: shreyas-goenka

bundle/config/experimental.go

@@ -42,6 +42,11 @@ type Experimental struct {
 	// Eventually this can be made the default once we have native CRUD in DABs
 	// at which point we can deprecate or remove this field all together.
 	SkipNamePrefixForSchema bool `json:"skip_name_prefix_for_schema,omitempty"`
+
+	// RecordDeploymentHistory opts the bundle into the deployment metadata
+	// service (DMS), which records deployment history and tracks what changed
+	// across deployments.
+	RecordDeploymentHistory bool `json:"record_deployment_history,omitempty"`
 }
 
 type Python struct {

bundle/deploy/lock/deployment_metadata_service.go

@@ -0,0 +1,204 @@
+package lock
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net/http"
+	"strconv"
+	"time"
+
+	clibundle "github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/internal/build"
+	"github.com/databricks/cli/libs/log"
+	"github.com/databricks/databricks-sdk-go/apierr"
+	sdkbundle "github.com/databricks/databricks-sdk-go/service/bundle"
+)
+
+const defaultHeartbeatInterval = 30 * time.Second
+
+// deploymentVersionRecorder records each deployment as a version with the
+// Deployment Metadata Service. It runs alongside the workspace-filesystem lock
+// inside DeploymentLock.Acquire / Release.
+//
+// The deployment ID is the state lineage (read from resources.json), so a
+// bundle deployment maps one-to-one to a DMS deployment record.
+type deploymentVersionRecorder struct {
+	b    *clibundle.Bundle
+	goal Goal
+
+	// populated by createVersion
+	svc           sdkbundle.BundleInterface
+	deploymentID  string
+	versionNum    int64
+	stopHeartbeat context.CancelFunc
+}
+
+func newDeploymentVersionRecorder(b *clibundle.Bundle, goal Goal) *deploymentVersionRecorder {
+	return &deploymentVersionRecorder{b: b, goal: goal}
+}
+
+func (r *deploymentVersionRecorder) createVersion(ctx context.Context) error {
+	versionType, ok := goalToVersionType(r.goal)
+	if !ok {
+		return fmt.Errorf("%s is not supported with the deployment metadata service", r.goal)
+	}
+
+	r.svc = r.b.WorkspaceClient(ctx).Bundle
+
+	// The deployment ID is the state lineage. GetOrInitLineage generates one on
+	// the first deploy and stores it so the deploy persists the same value.
+	r.deploymentID = r.b.DeploymentBundle.StateDB.GetOrInitLineage()
+
+	versionID, err := createDeploymentVersion(ctx, r.b, r.svc, r.deploymentID, versionType)
+	if err != nil {
+		return err
+	}
+
+	versionNum, err := strconv.ParseInt(versionID, 10, 64)
+	if err != nil {
+		return fmt.Errorf("failed to parse version ID %q: %w", versionID, err)
+	}
+	r.versionNum = versionNum
+	r.stopHeartbeat = startHeartbeat(ctx, r.svc, r.deploymentID, versionID)
+	return nil
+}
+
+func (r *deploymentVersionRecorder) completeVersion(ctx context.Context, status DeploymentStatus) error {
+	if r.stopHeartbeat != nil {
+		r.stopHeartbeat()
+	}
+
+	versionIDStr := strconv.FormatInt(r.versionNum, 10)
+	versionName := fmt.Sprintf("deployments/%s/versions/%s", r.deploymentID, versionIDStr)
+
+	reason := sdkbundle.VersionCompleteVersionCompleteSuccess
+	if status == DeploymentFailure {
+		reason = sdkbundle.VersionCompleteVersionCompleteFailure
+	}
+
+	_, err := r.svc.CompleteVersion(ctx, sdkbundle.CompleteVersionRequest{
+		Name:             versionName,
+		CompletionReason: reason,
+	})
+	if err != nil {
+		return err
+	}
+	log.Infof(ctx, "Completed deployment version: deployment=%s version=%s reason=%s", r.deploymentID, versionIDStr, reason)
+
+	// For destroy operations, delete the deployment record after the version
+	// completes successfully.
+	if status == DeploymentSuccess && r.goal == GoalDestroy {
+		err = r.svc.DeleteDeployment(ctx, sdkbundle.DeleteDeploymentRequest{
+			Name: "deployments/" + r.deploymentID,
+		})
+		if err != nil {
+			return fmt.Errorf("failed to delete deployment: %w", err)
+		}
+	}
+
+	return nil
+}
+
+// createDeploymentVersion ensures the deployment record exists, then creates a
+// new version. The deployment ID is the state lineage: we GetDeployment first
+// and only CreateDeployment when it does not exist yet.
+func createDeploymentVersion(ctx context.Context, b *clibundle.Bundle, svc sdkbundle.BundleInterface, deploymentID string, versionType sdkbundle.VersionType) (versionID string, err error) {
+	dep, getErr := svc.GetDeployment(ctx, sdkbundle.GetDeploymentRequest{
+		Name: "deployments/" + deploymentID,
+	})
+	switch {
+	case errors.Is(getErr, apierr.ErrNotFound):
+		// Fresh deployment: create the record and start at version 1.
+		_, createErr := svc.CreateDeployment(ctx, sdkbundle.CreateDeploymentRequest{
+			DeploymentId: deploymentID,
+			Deployment: sdkbundle.Deployment{
+				TargetName: b.Config.Bundle.Target,
+			},
+		})
+		if createErr != nil {
+			return "", fmt.Errorf("failed to create deployment: %w", createErr)
+		}
+		versionID = "1"
+	case getErr != nil:
+		return "", fmt.Errorf("failed to get deployment: %w", getErr)
+	default:
+		// Existing deployment: increment the last version to get the next one.
+		lastVersion, parseErr := strconv.ParseInt(dep.LastVersionId, 10, 64)
+		if parseErr != nil {
+			return "", fmt.Errorf("failed to parse last_version_id %q: %w", dep.LastVersionId, parseErr)
+		}
+		versionID = strconv.FormatInt(lastVersion+1, 10)
+	}
+
+	// The server validates that versionID equals last_version_id + 1 and returns
+	// ABORTED otherwise (e.g. a concurrent deploy already created this version).
+	version, versionErr := svc.CreateVersion(ctx, sdkbundle.CreateVersionRequest{
+		Parent:    "deployments/" + deploymentID,
+		VersionId: versionID,
+		Version: sdkbundle.Version{
+			CliVersion:  build.GetInfo().Version,
+			VersionType: versionType,
+			TargetName:  b.Config.Bundle.Target,
+		},
+	})
+	if versionErr != nil {
+		return "", fmt.Errorf("failed to create deployment version: %w", versionErr)
+	}
+
+	log.Infof(ctx, "Created deployment version: deployment=%s version=%s", deploymentID, version.VersionId)
+	return versionID, nil
+}
+
+// startHeartbeat starts a background goroutine that sends heartbeats to keep
+// the deployment version's lease alive. Returns a cancel function to stop it.
+func startHeartbeat(ctx context.Context, svc sdkbundle.BundleInterface, deploymentID, versionID string) context.CancelFunc {
+	ctx, cancel := context.WithCancel(ctx)
+	versionName := fmt.Sprintf("deployments/%s/versions/%s", deploymentID, versionID)
+
+	go func() {
+		ticker := time.NewTicker(defaultHeartbeatInterval)
+		defer ticker.Stop()
+
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+				_, err := svc.Heartbeat(ctx, sdkbundle.HeartbeatRequest{Name: versionName})
+				if err != nil {
+					// A 409 ABORTED is expected if the version was completed
+					// between the ticker firing and the heartbeat.
+					if isAbortedErr(err) {
+						log.Debugf(ctx, "Heartbeat stopped: version already completed")
+						return
+					}
+					log.Warnf(ctx, "Failed to send deployment heartbeat: %v", err)
+				} else {
+					log.Debugf(ctx, "Deployment heartbeat sent: deployment=%s version=%s", deploymentID, versionID)
+				}
+			}
+		}
+	}()
+
+	return cancel
+}
+
+// isAbortedErr reports whether err is an HTTP 409 ABORTED from the DMS API.
+func isAbortedErr(err error) bool {
+	var apiErr *apierr.APIError
+	return errors.As(err, &apiErr) && apiErr.StatusCode == http.StatusConflict && apiErr.ErrorCode == "ABORTED"
+}
+
+// goalToVersionType maps a deployment goal to a DMS VersionType.
+// Returns false for goals not supported by the metadata service (bind/unbind).
+func goalToVersionType(goal Goal) (sdkbundle.VersionType, bool) {
+	switch goal {
+	case GoalDeploy:
+		return sdkbundle.VersionTypeVersionTypeDeploy, true
+	case GoalDestroy:
+		return sdkbundle.VersionTypeVersionTypeDestroy, true
+	default:
+		return "", false
+	}
+}

bundle/deploy/lock/lock.go

@@ -27,10 +27,14 @@ const (
 )
 
 // 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.
+// The workspace-filesystem lock always serializes concurrent deployments.
+// DMS version tracking is additive: CreateVersion is called after acquiring
+// the file lock, CompleteVersion before releasing it.
 type DeploymentLock struct {
 	wfs workspaceFilesystemLock
+	// dms records deployment versions with the metadata service. It is nil
+	// unless experimental.record_deployment_history is set.
+	dms *deploymentVersionRecorder
 }
 
 // NewDeploymentLock returns a DeploymentLock for the bundle.
@@ -54,15 +58,37 @@ func NewDeploymentLock(ctx context.Context, b *bundle.Bundle, goal Goal) *Deploy
 	if enabled {
 		l.wfs.client = b.WorkspaceClient(ctx)
 	}
+	if b.Config.Experimental != nil && b.Config.Experimental.RecordDeploymentHistory {
+		l.dms = newDeploymentVersionRecorder(b, goal)
+	}
 	return l
 }
 
 // Acquire acquires the deployment lock.
+// The workspace-filesystem lock is always acquired first.
+// When DMS recording is enabled, a new deployment version is created after the
+// file lock so the deployment is also registered server-side.
+//
+// Optimization: once the deployment is tracked by DMS (the state lineage has a
+// server-side record), the file lock could be skipped — creating a version
+// provides equivalent concurrency control via the server-side version counter.
 func (l *DeploymentLock) Acquire(ctx context.Context) error {
-	return l.wfs.acquire(ctx)
+	if err := l.wfs.acquire(ctx); err != nil {
+		return err
+	}
+	if l.dms != nil {
+		return l.dms.createVersion(ctx)
+	}
+	return nil
 }
 
 // Release releases the deployment lock.
+// When DMS is enabled, CompleteVersion is called before releasing the file lock.
 func (l *DeploymentLock) Release(ctx context.Context, status DeploymentStatus) error {
+	if l.dms != nil {
+		if err := l.dms.completeVersion(ctx, status); err != nil {
+			return err
+		}
+	}
 	return l.wfs.release(ctx, status)
 }

bundle/direct/dstate/state.go

@@ -147,6 +147,22 @@ func (db *DeploymentState) GetResourceID(key string) string {
 	return db.stateIDs[key]
 }
 
+// GetOrInitLineage returns the state lineage, generating and storing a new one
+// in memory if the state does not have a lineage yet (fresh deployment). Storing
+// it in Data.Lineage means a subsequent Open-for-write / UpgradeToWrite persists
+// the same value, so the lineage the caller observes here matches the one written
+// to the state file by the deploy. This lets the DMS lock derive a stable
+// deployment ID from the lineage even on the first deploy.
+func (db *DeploymentState) GetOrInitLineage() string {
+	db.mu.Lock()
+	defer db.mu.Unlock()
+
+	if db.Data.Lineage == "" {
+		db.Data.Lineage = uuid.New().String()
+	}
+	return db.Data.Lineage
+}
+
 type (
 	// If true, then Open reads the WAL and merges it in the state. If false, and WAL is present, Open returns an error.
 	WithRecovery bool

bundle/internal/schema/annotations.yml

@@ -75,6 +75,9 @@ github.com/databricks/cli/bundle/config.Experimental:
   "python_wheel_wrapper":
     "description": |-
       Whether to use a Python wheel wrapper.
+  "record_deployment_history":
+    "description": |-
+      Whether to record deployment history using the deployment metadata service (DMS), which tracks what changed across deployments.
   "scripts":
     "description": |-
       The commands to run.