diff --git a/acceptance/bundle/state/feature_flags/databricks.yml b/acceptance/bundle/state/feature_flags/databricks.yml new file mode 100644 index 00000000000..5134dbcc12c --- /dev/null +++ b/acceptance/bundle/state/feature_flags/databricks.yml @@ -0,0 +1,7 @@ +bundle: + name: test-bundle + +resources: + jobs: + my_job: + name: "my job" diff --git a/acceptance/bundle/state/feature_flags/out.test.toml b/acceptance/bundle/state/feature_flags/out.test.toml new file mode 100644 index 00000000000..e90b6d5d1ba --- /dev/null +++ b/acceptance/bundle/state/feature_flags/out.test.toml @@ -0,0 +1,3 @@ +Local = true +Cloud = false +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] diff --git a/acceptance/bundle/state/feature_flags/output.txt b/acceptance/bundle/state/feature_flags/output.txt new file mode 100644 index 00000000000..e9279ea54f7 --- /dev/null +++ b/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 diff --git a/acceptance/bundle/state/feature_flags/resources.empty_features.json b/acceptance/bundle/state/feature_flags/resources.empty_features.json new file mode 100644 index 00000000000..b20c97aa074 --- /dev/null +++ b/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": {} +} diff --git a/acceptance/bundle/state/feature_flags/resources.known.json b/acceptance/bundle/state/feature_flags/resources.known.json new file mode 100644 index 00000000000..19caabaf2a0 --- /dev/null +++ b/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": {} +} diff --git a/acceptance/bundle/state/feature_flags/resources.unknown.json b/acceptance/bundle/state/feature_flags/resources.unknown.json new file mode 100644 index 00000000000..b844b098a70 --- /dev/null +++ b/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": {} +} diff --git a/acceptance/bundle/state/feature_flags/script b/acceptance/bundle/state/feature_flags/script new file mode 100644 index 00000000000..5d0f8f57693 --- /dev/null +++ b/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:" diff --git a/acceptance/bundle/state/feature_flags/test.toml b/acceptance/bundle/state/feature_flags/test.toml new file mode 100644 index 00000000000..7fc493d51a4 --- /dev/null +++ b/acceptance/bundle/state/feature_flags/test.toml @@ -0,0 +1,4 @@ +Ignore = [".databricks"] + +[EnvMatrix] +DATABRICKS_BUNDLE_ENGINE = ["direct"] diff --git a/acceptance/bundle/state/future_version/output.txt b/acceptance/bundle/state/future_version/output.txt index 7cf98129ee9..0a16971f472 100644 --- a/acceptance/bundle/state/future_version/output.txt +++ b/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 diff --git a/bundle/direct/dstate/migrate.go b/bundle/direct/dstate/migrate.go index 381d63a12eb..b105fd42003 100644 --- a/bundle/direct/dstate/migrate.go +++ b/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++ { diff --git a/bundle/direct/dstate/state.go b/bundle/direct/dstate/state.go index eb41217f61d..02853fb6135 100644 --- a/bundle/direct/dstate/state.go +++ b/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) diff --git a/bundle/direct/dstate/state_test.go b/bundle/direct/dstate/state_test.go index b52f2bf4d82..87b183a708b 100644 --- a/bundle/direct/dstate/state_test.go +++ b/bundle/direct/dstate/state_test.go @@ -90,6 +90,101 @@ func TestHeaderOnlyWALRecoveryDoesNotAdvanceSerial(t *testing.T) { mustFinalize(t, &recovered) } +func TestOpenWithFeaturePersists(t *testing.T) { + path := filepath.Join(t.TempDir(), "state.json") + + // Opting into a feature at Open records it; the subsequent write persists it. + var db DeploymentState + require.NoError(t, db.Open(t.Context(), path, WithRecovery(true), WithWrite(true), featureDummy)) + require.NoError(t, db.SaveState("jobs.my_job", "123", map[string]string{"key": "val"}, nil)) + mustFinalize(t, &db) + + // Recording a feature writes featureStateVersion so older CLIs refuse the state. + var db2 DeploymentState + require.NoError(t, db2.Open(t.Context(), path, WithRecovery(false), WithWrite(false))) + assert.Equal(t, featureStateVersion, db2.Data.StateVersion) + assert.Equal(t, map[string]FeatureInfo{featureDummy: {}}, db2.Data.Features) + mustFinalize(t, &db2) +} + +// TestEmptyFeatureStateEquivalentToVersion2 documents and pins the special case +// that a version-3 state with no features is read as version 2. This is scaffolding +// for the deferred version bump, special-cased to version 3 only (see the +// featureStateVersion doc comment). +// +// When the baseline is actually bumped to 3, this equivalence must go away — a real +// version-3 state must not be reinterpreted as 2. This test is the forcing function: +// it fails once featureStateVersion is removed or the normalization changes, making +// the author decide what the post-bump behavior should be. +func TestEmptyFeatureStateEquivalentToVersion2(t *testing.T) { + // The special case applies to featureStateVersion (3) only, and only when the + // feature map is empty; a populated feature map keeps the version as-is. + require.Equal(t, 2, currentStateVersion, "when currentStateVersion is bumped, remove featureStateVersion and this special case") + require.Equal(t, 3, featureStateVersion) + + empty := &Database{Header: Header{StateVersion: featureStateVersion}} + require.NoError(t, migrateState(empty)) + assert.Equal(t, currentStateVersion, empty.StateVersion, "v3 + no features is read as v2") + + withFeature := &Database{Header: Header{ + StateVersion: featureStateVersion, + Features: map[string]FeatureInfo{featureDummy: {}}, + }} + require.NoError(t, migrateState(withFeature)) + assert.Equal(t, featureStateVersion, withFeature.StateVersion, "v3 + features stays v3") +} + +func TestRecoveryPreservesFeatures(t *testing.T) { + path := filepath.Join(t.TempDir(), "state.json") + + // Commit an initial state with no features. + var db DeploymentState + require.NoError(t, db.Open(t.Context(), path, WithRecovery(true), WithWrite(true))) + require.NoError(t, db.SaveState("jobs.my_job", "1", map[string]string{}, nil)) + mustFinalize(t, &db) + + // A feature-recording deploy writes WAL entries but crashes before Finalize, + // leaving the WAL behind. Closing walFile without finalizing simulates the crash. + var crashed DeploymentState + require.NoError(t, crashed.Open(t.Context(), path, WithRecovery(true), WithWrite(true), featureDummy)) + require.NoError(t, crashed.SaveState("jobs.my_job", "1", map[string]string{"changed": "yes"}, nil)) + require.NoError(t, crashed.walFile.Close()) + + // Recovery must carry the feature (and its version) forward from the WAL header, + // not silently drop it back to the pre-crash featureless state. + var recovered DeploymentState + require.NoError(t, recovered.Open(t.Context(), path, WithRecovery(true), WithWrite(false))) + assert.Equal(t, featureStateVersion, recovered.Data.StateVersion) + assert.Equal(t, map[string]FeatureInfo{featureDummy: {}}, recovered.Data.Features) + mustFinalize(t, &recovered) +} + +func TestOpenWithUnknownFeaturePanics(t *testing.T) { + path := filepath.Join(t.TempDir(), "state.json") + + var db DeploymentState + assert.Panics(t, func() { + _ = db.Open(t.Context(), path, WithRecovery(true), WithWrite(true), "no-such-feature") + }) +} + +func TestCheckSupportedFeatures(t *testing.T) { + // Known features (and none at all) are accepted; an unknown feature is rejected + // with its name and the fixed features documentation link. + require.NoError(t, checkSupportedFeatures(&Database{})) + require.NoError(t, checkSupportedFeatures(&Database{Header: Header{ + Features: map[string]FeatureInfo{featureDummy: {}}, + }})) + + err := checkSupportedFeatures(&Database{Header: Header{ + Features: map[string]FeatureInfo{"future_feature": {}}, + }}) + require.Error(t, err) + assert.Contains(t, err.Error(), `feature "future_feature"`) + assert.Contains(t, err.Error(), "upgrade to the latest CLI version") + assert.Contains(t, err.Error(), featuresDocURL) +} + func TestDeleteState(t *testing.T) { path := filepath.Join(t.TempDir(), "state.json")