bundle/direct: add a feature-flag list to the direct deployment state

acceptance/bundle/state/feature_flags/databricks.yml

@@ -0,0 +1,7 @@
+bundle:
+  name: test-bundle
+
+resources:
+  jobs:
+    my_job:
+      name: "my job"

acceptance/bundle/state/feature_flags/output.txt

@@ -0,0 +1,19 @@
+
+=== a state recording a known feature flag is accepted
+>>> [CLI] bundle plan
+create jobs.my_job
+
+Plan: 1 to add, 0 to change, 0 to delete, 0 unchanged
+
+=== a state recording an unknown feature flag is rejected
+>>> errcode [CLI] bundle plan
+Error: the deployment state requires feature "future_feature" which this CLI does not support; upgrade to the latest CLI version and see https://docs.databricks.com/aws/en/dev-tools/bundles/state-features#features for more information
+
+
+Exit code: 1
+
+=== a version-3 state with an empty features map is accepted (treated as version 2)
+>>> [CLI] bundle plan
+create jobs.my_job
+
+Plan: 1 to add, 0 to change, 0 to delete, 0 unchanged

acceptance/bundle/state/feature_flags/resources.empty_features.json

@@ -0,0 +1,8 @@
+{
+ "state_version": 3,
+ "features": {},
+ "cli_version": "0.0.0-dev",
+ "lineage": "test-lineage",
+ "serial": 1,
+ "state": {}
+}

acceptance/bundle/state/feature_flags/resources.known.json

@@ -0,0 +1,10 @@
+{
+ "state_version": 3,
+ "features": {
+  "dummy": {}
+ },
+ "cli_version": "0.0.0-dev",
+ "lineage": "test-lineage",
+ "serial": 1,
+ "state": {}
+}

acceptance/bundle/state/feature_flags/resources.unknown.json

@@ -0,0 +1,10 @@
+{
+ "state_version": 3,
+ "features": {
+  "future_feature": {}
+ },
+ "cli_version": "0.0.0-dev",
+ "lineage": "test-lineage",
+ "serial": 1,
+ "state": {}
+}

acceptance/bundle/state/feature_flags/script

@@ -0,0 +1,13 @@
+mkdir -p .databricks/bundle/default
+
+title "a state recording a known feature flag is accepted"
+cp resources.known.json .databricks/bundle/default/resources.json
+trace $CLI bundle plan | contains.py "Plan:"
+
+title "a state recording an unknown feature flag is rejected"
+cp resources.unknown.json .databricks/bundle/default/resources.json
+trace errcode $CLI bundle plan 2>&1 | contains.py 'requires feature "future_feature"' "upgrade to the latest CLI version" "https://docs.databricks.com/aws/en/dev-tools/bundles/state-features#features"
+
+title "a version-3 state with an empty features map is accepted (treated as version 2)"
+cp resources.empty_features.json .databricks/bundle/default/resources.json
+trace $CLI bundle plan | contains.py "Plan:"

acceptance/bundle/state/feature_flags/test.toml

@@ -0,0 +1,4 @@
+Ignore = [".databricks"]
+
+[EnvMatrix]
+DATABRICKS_BUNDLE_ENGINE = ["direct"]

acceptance/bundle/state/future_version/output.txt

@@ -1,3 +1,3 @@
-state version 999 is newer than supported version 2; upgrade the CLI
+state version 999 is newer than supported version 3; upgrade the CLI
 
 Exit code: 1

bundle/direct/dstate/migrate.go

@@ -13,11 +13,23 @@ import (
 // migrateState runs all necessary migrations on the database.
 // It is called after loading state from disk.
 func migrateState(db *Database) error {
-	if db.StateVersion == currentStateVersion {
+	// A featureStateVersion state with no features is equivalent to
+	// currentStateVersion; normalize it so this CLI treats it as current rather
+	// than rejecting it as too new. This lets a later release write
+	// featureStateVersion unconditionally without breaking CLIs with this change.
+	if db.StateVersion == featureStateVersion && len(db.Features) == 0 {
+		db.StateVersion = currentStateVersion
+	}
+
+	// featureStateVersion is the newest version this CLI reads. A featureStateVersion
+	// state carries no structural change over currentStateVersion (only the feature
+	// list, validated separately by checkSupportedFeatures), so there is nothing to
+	// migrate. Anything past it was written by a newer CLI and is refused.
+	if db.StateVersion == currentStateVersion || db.StateVersion == featureStateVersion {
 		return nil
 	}
-	if db.StateVersion > currentStateVersion {
-		return fmt.Errorf("state version %d is newer than supported version %d; upgrade the CLI", db.StateVersion, currentStateVersion)
+	if db.StateVersion > featureStateVersion {
+		return fmt.Errorf("state version %d is newer than supported version %d; upgrade the CLI", db.StateVersion, featureStateVersion)
 	}
 
 	for version := db.StateVersion; version < currentStateVersion; version++ {

bundle/direct/dstate/state.go

@@ -10,6 +10,7 @@ import (
 	"io/fs"
 	"os"
 	"path/filepath"
+	"slices"
 	"strings"
 	"sync"
 
@@ -21,12 +22,59 @@ import (
 )
 
 const (
+	// currentStateVersion is the schema version written for deployments that record
+	// no feature flags, and the version legacy states are migrated up to on load.
 	currentStateVersion = 2
-	initialBufferSize   = 64 * 1024
-	maxWalEntrySize     = 10 * 1024 * 1024
-	walSuffix           = ".wal"
+
+	// featureStateVersion is written for states that record feature flags (see
+	// Header.Features) and is the newest version this CLI can read. Writing it (not
+	// currentStateVersion) makes older CLIs, which read only up to
+	// currentStateVersion, refuse a state that depends on a feature they lack.
+	//
+	// A featureStateVersion state with no features is treated as currentStateVersion
+	// (see migrateState). That lets a later release start writing featureStateVersion
+	// unconditionally — "the version bump" — while CLIs with this change already
+	// tolerate such states instead of rejecting them.
+	//
+	// IMPORTANT: this "version 3 + empty features == version 2" equivalence is
+	// special-cased to version 3 only. It is scaffolding for the deferred bump, not
+	// a general rule. When you actually bump the baseline (currentStateVersion) to 3,
+	// delete featureStateVersion, delete the normalization branch in migrateState,
+	// and delete TestEmptyFeatureStateEquivalentToVersion2 — otherwise a real v3
+	// state gets silently reinterpreted as v2. The test exists to force this cleanup.
+	featureStateVersion = 3
+
+	initialBufferSize = 64 * 1024
+	maxWalEntrySize   = 10 * 1024 * 1024
+	walSuffix         = ".wal"
 )
 
+// featuresDocURL is the single documentation page describing deployment state
+// feature flags. It is shown when a state records a feature this CLI does not
+// support; it is a fixed link for all features, not derived per feature. The
+// #features anchor points at the feature table; if it ever breaks, the user still
+// lands on the page.
+const featuresDocURL = "https://docs.databricks.com/aws/en/dev-tools/bundles/state-features#features"
+
+// FeatureInfo is the per-feature value recorded in the state. It is intentionally
+// empty: features are keyed by name and the presence of the key is what matters.
+// It is a struct (rather than making Features a set/list) so per-feature fields
+// can be added later without changing the state's shape.
+type FeatureInfo struct{}
+
+// featureDummy is a placeholder feature flag used only to exercise the mechanism
+// in tests; no real deployment ever records it.
+const featureDummy = "dummy"
+
+// supportedFeatures is the set of deployment feature flags this CLI supports. To
+// define a feature flag, add its name here (and document it at featuresDocURL).
+//
+// A state recording a feature absent from this set was written by a newer CLI;
+// checkSupportedFeatures rejects it and points the user at featuresDocURL.
+var supportedFeatures = map[string]struct{}{
+	featureDummy: {},
+}
+
 // errStaleWAL is returned when the WAL serial is behind the expected serial.
 // The caller should delete the stale WAL and proceed normally.
 var errStaleWAL = errors.New("stale WAL")
@@ -42,10 +90,17 @@ type DeploymentState struct {
 }
 
 type Header struct {
-	StateVersion int    `json:"state_version"`
-	CLIVersion   string `json:"cli_version"`
-	Lineage      string `json:"lineage"`
-	Serial       int    `json:"serial"`
+	StateVersion int `json:"state_version"`
+
+	// Features maps each feature flag this state depends on to a (currently empty)
+	// FeatureInfo. A CLI that does not recognize a feature refuses the state and
+	// points the user at the feature's documentation (see checkSupportedFeatures).
+	// Empty/omitted for states that use no features.
+	Features map[string]FeatureInfo `json:"features,omitempty"`
+
+	CLIVersion string `json:"cli_version"`
+	Lineage    string `json:"lineage"`
+	Serial     int    `json:"serial"`
 }
 
 type Database struct {
@@ -167,6 +222,43 @@ func (db *DeploymentState) GetOrInitLineage() string {
 	return db.Data.Lineage
 }
 
+// recordFeatures stamps the given opted-in features into the state, so a later
+// write persists them and older CLIs that don't support them are turned away.
+// Each name must be a known feature (the caller opts in by passing it to Open).
+func (db *Database) recordFeatures(names []string) {
+	for _, name := range names {
+		if _, ok := supportedFeatures[name]; !ok {
+			panic(fmt.Sprintf("internal error: unknown feature %q", name))
+		}
+		if db.Features == nil {
+			db.Features = make(map[string]FeatureInfo)
+		}
+		db.Features[name] = FeatureInfo{}
+	}
+	// A state that records a feature must be written at featureStateVersion so an
+	// older CLI, which reads only up to currentStateVersion, refuses it rather than
+	// silently ignoring the feature it depends on.
+	if len(db.Features) > 0 {
+		db.StateVersion = featureStateVersion
+	}
+}
+
+// checkSupportedFeatures rejects a state that records a feature flag this CLI does
+// not understand, pointing the user at the feature's documentation.
+func checkSupportedFeatures(db *Database) error {
+	names := make([]string, 0, len(db.Features))
+	for name := range db.Features {
+		names = append(names, name)
+	}
+	slices.Sort(names)
+	for _, name := range names {
+		if _, ok := supportedFeatures[name]; !ok {
+			return fmt.Errorf("the deployment state requires feature %q which this CLI does not support; upgrade to the latest CLI version and see %s for more information", name, featuresDocURL)
+		}
+	}
+	return nil
+}
+
 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
@@ -176,7 +268,7 @@ type (
 	WithWrite bool
 )
 
-func (db *DeploymentState) Open(ctx context.Context, path string, withRecovery WithRecovery, withWrite WithWrite) error {
+func (db *DeploymentState) Open(ctx context.Context, path string, withRecovery WithRecovery, withWrite WithWrite, features ...string) error {
 	db.mu.Lock()
 	defer db.mu.Unlock()
 
@@ -224,6 +316,12 @@ func (db *DeploymentState) Open(ctx context.Context, path string, withRecovery W
 		return fmt.Errorf("migrating state %s: %w", path, err)
 	}
 
+	if err := checkSupportedFeatures(&db.Data); err != nil {
+		return err
+	}
+
+	db.Data.recordFeatures(features)
+
 	if withWrite {
 		if err := os.MkdirAll(filepath.Dir(walPath), 0o755); err != nil {
 			return fmt.Errorf("failed to create state directory: %w", err)
@@ -236,7 +334,8 @@ func (db *DeploymentState) Open(ctx context.Context, path string, withRecovery W
 		walHead := Header{
 			Lineage:      db.GetOrInitLineage(),
 			Serial:       db.Data.Serial + 1,
-			StateVersion: currentStateVersion,
+			StateVersion: db.Data.StateVersion,
+			Features:     db.Data.Features,
 			CLIVersion:   build.GetInfo().Version,
 		}
 		return appendJSONLine(db.walFile, walHead)
@@ -302,6 +401,7 @@ func (db *DeploymentState) mergeWalIntoState(ctx context.Context) (bool, error)
 	lineNumber := 0
 	var corruptedLines [][]byte
 	var newSerial int
+	var newHeader Header
 
 	for scanner.Scan() {
 		lineNumber++
@@ -326,6 +426,7 @@ func (db *DeploymentState) mergeWalIntoState(ctx context.Context) (bool, error)
 				return false, fmt.Errorf("WAL serial (%d) is ahead of expected (%d), state may be corrupted", header.Serial, expectedSerial)
 			}
 			newSerial = header.Serial
+			newHeader = header
 		} else {
 			var entry WALEntry
 			if err := json.Unmarshal(line, &entry); err != nil {
@@ -370,6 +471,11 @@ func (db *DeploymentState) mergeWalIntoState(ctx context.Context) (bool, error)
 	// "ahead of expected". See acceptance/bundle/deploy/wal/header-only-wal.
 	if hasEntries {
 		db.Data.Serial = newSerial
+		// Restore the schema version and feature flags the interrupted deploy was
+		// writing, so a crash between the WAL header and Finalize does not drop the
+		// features the recovered state depends on.
+		db.Data.StateVersion = newHeader.StateVersion
+		db.Data.Features = newHeader.Features
 	}
 
 	return hasEntries, nil
@@ -432,7 +538,8 @@ func (db *DeploymentState) UpgradeToWrite() error {
 	walHead := Header{
 		Lineage:      db.GetOrInitLineage(),
 		Serial:       db.Data.Serial + 1,
-		StateVersion: currentStateVersion,
+		StateVersion: db.Data.StateVersion,
+		Features:     db.Data.Features,
 		CLIVersion:   build.GetInfo().Version,
 	}
 	return appendJSONLine(db.walFile, walHead)