diff --git a/bundle/direct/bind.go b/bundle/direct/bind.go index 7fbf7ede662..74b04038342 100644 --- a/bundle/direct/bind.go +++ b/bundle/direct/bind.go @@ -2,6 +2,7 @@ package direct import ( "context" + "errors" "fmt" "os" "strings" @@ -63,13 +64,16 @@ type BindResult struct { func (b *DeploymentBundle) Bind(ctx context.Context, client *databricks.WorkspaceClient, configRoot *config.Root, statePath, resourceKey, resourceID string) (*BindResult, error) { // Check if the resource is already managed (bound to a different ID) var checkStateDB dstate.DeploymentState - if err := checkStateDB.Open(ctx, statePath, dstate.ReadingFile, dstate.WithRecovery(true), dstate.OpenDmsArgs{}); err != nil { + if err := checkStateDB.Open(ctx, statePath, dstate.WithRecovery(true)); err != nil { // State that cannot be read is not the same as state without a binding: // the resource may well be managed already, and binding on top of it // would take over a resource whose ownership was never checked. A state // file that does not exist yet opens successfully as an empty one. return nil, fmt.Errorf("cannot check whether %s is already bound: %w", resourceKey, err) } + if checkStateDB.IsDeploymentMetadataService() { + return nil, errors.New("bind is not supported for a bundle target that records deployment history") + } existingID := checkStateDB.GetResourceID(resourceKey) if _, err := checkStateDB.Finalize(ctx); err != nil { log.Warnf(ctx, "failed to finalize state: %v", err) @@ -92,7 +96,7 @@ func (b *DeploymentBundle) Bind(ctx context.Context, client *databricks.Workspac } // Open temp state - err := b.StateDB.Open(ctx, tmpStatePath, dstate.ReadingFile, dstate.WithRecovery(false), dstate.OpenDmsArgs{}) + err := b.StateDB.Open(ctx, tmpStatePath, dstate.WithRecovery(false)) if err != nil { os.Remove(tmpStatePath) return nil, err @@ -119,7 +123,7 @@ func (b *DeploymentBundle) Bind(ctx context.Context, client *databricks.Workspac log.Infof(ctx, "Bound %s to id=%s (in temp state)", resourceKey, resourceID) // First plan + update: populate state with resolved config - err = b.StateDB.Open(ctx, tmpStatePath, dstate.ReadingFile, dstate.WithRecovery(true), dstate.OpenDmsArgs{}) + err = b.StateDB.Open(ctx, tmpStatePath, dstate.WithRecovery(true)) if err != nil { os.Remove(tmpStatePath) return nil, err @@ -168,7 +172,7 @@ func (b *DeploymentBundle) Bind(ctx context.Context, client *databricks.Workspac return nil, fmt.Errorf("compacting state: %w", err) } - err = b.StateDB.Open(ctx, tmpStatePath, dstate.ReadingFile, dstate.WithRecovery(true), dstate.OpenDmsArgs{}) + err = b.StateDB.Open(ctx, tmpStatePath, dstate.WithRecovery(true)) if err != nil { os.Remove(tmpStatePath) return nil, err @@ -192,7 +196,7 @@ func (b *DeploymentBundle) Bind(ctx context.Context, client *databricks.Workspac } // Second plan: this is the plan to present to the user (change between remote resource and config) - err = b.StateDB.Open(ctx, tmpStatePath, dstate.ReadingFile, dstate.WithRecovery(true), dstate.OpenDmsArgs{}) + err = b.StateDB.Open(ctx, tmpStatePath, dstate.WithRecovery(true)) if err != nil { os.Remove(tmpStatePath) return nil, err @@ -242,9 +246,12 @@ func (result *BindResult) Cancel() { // Unbind removes a resource from direct engine state without deleting // the workspace resource. Also removes associated permissions/grants entries. func (b *DeploymentBundle) Unbind(ctx context.Context, statePath, resourceKey string) error { - if err := b.StateDB.Open(ctx, statePath, dstate.ReadingFile, dstate.WithRecovery(true), dstate.OpenDmsArgs{}); err != nil { + if err := b.StateDB.Open(ctx, statePath, dstate.WithRecovery(true)); err != nil { return err } + if b.StateDB.IsDeploymentMetadataService() { + return errors.New("unbind is not supported for a bundle target that records deployment history") + } if err := b.StateDB.UpgradeToWrite(); err != nil { return err } diff --git a/bundle/direct/bind_test.go b/bundle/direct/bind_test.go new file mode 100644 index 00000000000..2a9adc30ed3 --- /dev/null +++ b/bundle/direct/bind_test.go @@ -0,0 +1,38 @@ +package direct + +import ( + "encoding/json" + "os" + "path/filepath" + "testing" + + "github.com/databricks/cli/bundle/direct/dstate" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestBindAndUnbindRejectDMSAuthoritativeState(t *testing.T) { + for _, operation := range []string{"bind", "unbind"} { + t.Run(operation, func(t *testing.T) { + path := filepath.Join(t.TempDir(), "resources.json") + state := dstate.NewDatabase("lineage", 1) + state.Features = map[string]struct{}{dstate.FeatureDeploymentHistory: {}} + data, err := json.Marshal(state) + require.NoError(t, err) + require.NoError(t, os.WriteFile(path, data, 0o600)) + + var db DeploymentBundle + // Reject file-only operations before they need a resource API client. + if operation == "bind" { + _, err = db.Bind(t.Context(), nil, nil, path, "resources.jobs.foo", "123") + } else { + err = db.Unbind(t.Context(), path, "resources.jobs.foo") + } + require.ErrorContains(t, err, operation+" is not supported") + assert.NoFileExists(t, path+".wal") + persisted, err := os.ReadFile(path) + require.NoError(t, err) + assert.Equal(t, data, persisted) + }) + } +} diff --git a/bundle/direct/dstate/dms.go b/bundle/direct/dstate/dms.go index 57e10b8d4b4..de26fe9a905 100644 --- a/bundle/direct/dstate/dms.go +++ b/bundle/direct/dstate/dms.go @@ -1,11 +1,15 @@ package dstate import ( + "context" "encoding/json" "fmt" "github.com/databricks/cli/bundle/deployplan" + "github.com/databricks/cli/libs/cmdctx" "github.com/databricks/cli/libs/dms" + "github.com/databricks/cli/libs/retry" + "github.com/databricks/databricks-sdk-go/service/bundledeployments" ) // RecordedState is what the CLI serializes into the DMS operation's state field. It wraps the @@ -16,6 +20,43 @@ type RecordedState struct { DependsOn []deployplan.DependsOnEntry `json:"depends_on,omitempty"` } +// LoadDMSState loads authoritative resources into an already-open read-only checkpoint. +// The service version must be at least as recent as the file's cached serial. +func (db *DeploymentState) LoadDMSState(ctx context.Context, deploymentID string, lastVersionID int) error { + db.mu.Lock() + defer db.mu.Unlock() + db.AssertOpenedForRead() + + if db.Data.Serial > lastVersionID { + return fmt.Errorf("inconsistent deployment state: resource state serial (%d) is ahead of deployment history version (%d)", db.Data.Serial, lastVersionID) + } + if deploymentID != "" { + w := cmdctx.WorkspaceClient(ctx) + recorded, err := retry.With(ctx, retry.IsTransientRead, func() ([]bundledeployments.Resource, error) { + return w.BundleDeployments.ListResourcesAll(ctx, bundledeployments.ListResourcesRequest{ + Parent: dms.DeploymentName(deploymentID), + }) + }) + if err != nil { + return fmt.Errorf("listing resources from the deployment history service: %w", err) + } + resources := make([]dms.Resource, len(recorded)) + for i, res := range recorded { + // DMS keys drop the "resources." prefix; add it back for local state. + resources[i] = dms.Resource{ + Key: dms.StatePrefix + res.ResourceKey, + ID: res.ResourceId, + State: res.State, + } + } + if err := db.applyDMSState(resources); err != nil { + return err + } + } + db.DeploymentID, db.VersionID = deploymentID, lastVersionID + return nil +} + // applyDMSState fills in the resource state from what DMS recorded. For a deployment that uses DMS // resources.json does not track state. The service is queried for state for resources and then they // are filled in-place in DeploymentState.Data.State: diff --git a/bundle/direct/dstate/state.go b/bundle/direct/dstate/state.go index 384246913f4..ec33ee0029e 100644 --- a/bundle/direct/dstate/state.go +++ b/bundle/direct/dstate/state.go @@ -24,7 +24,6 @@ import ( "github.com/databricks/cli/libs/cmdctx" "github.com/databricks/cli/libs/dms" "github.com/databricks/cli/libs/log" - "github.com/databricks/cli/libs/retry" "github.com/databricks/databricks-sdk-go/service/bundledeployments" ) @@ -82,11 +81,6 @@ func assertNoUnsupportedFeatures(features map[string]struct{}) error { // The caller should delete the stale WAL and proceed normally. var errStaleWAL = errors.New("stale WAL") -// ErrUnsettingRecording is returned by Open when an operation that does not support deployment -// history (such as unbind) reaches a recorded state. Callers present an operation-appropriate -// message via errors.Is. -var ErrUnsettingRecording = errors.New("this operation is not supported for a deployment that records deployment history") - type DeploymentState struct { Path string Data Database @@ -111,29 +105,16 @@ type DeploymentState struct { // completion after an explicit one does nothing. versionCompleted bool - // DeploymentID is the recorded deployment's id. Set at Open, or by the first deploy that creates - // the deployment, which Open cannot know about. + // DeploymentID identifies the history deployment. Set when loading history metadata + // or when the first deploy creates it. DeploymentID string // VersionID is the service-owned counter for recorded deployments. A copy is stored in // resources.json as Serial to ensure outdated resource state is not read. - // Open always takes the authoritative value from the service. + // History preparation loads the authoritative value from the service. VersionID int } -// OpenDmsArgs identifies the recorded deployment Open reads from. The zero value means the -// bundle does not record deployment history, or no deployment exists for it yet. -type OpenDmsArgs struct { - // DeploymentID is the deployment's server-minted id. - DeploymentID string - - // LastVersionID is the most recent version the service has recorded, zero when it has none. - // Open takes the state's version from it: the service owns the number. - // - // Equivalent to serial for a file based deployment that uses resources.json to track state, - LastVersionID int -} - type Header struct { StateVersion int `json:"state_version"` @@ -463,6 +444,10 @@ func (db *DeploymentState) GetSerial() int { // isDeploymentMetadataService is IsDeploymentMetadataService for callers already holding db.mu. func (db *DeploymentState) isDeploymentMetadataService() bool { + if db.mode == ReadingFile { + _, only := db.Data.Features[FeatureDeploymentHistory] + return only + } return db.mode == WritingDeploymentHistoryOnly } @@ -481,8 +466,8 @@ func (db *DeploymentState) RecordsHistory() bool { defer db.mu.Unlock() switch db.mode { case ReadingFile: - _, history := db.Data.Features[FeatureDeploymentHistoryShadow] - return history + _, shadow := db.Data.Features[FeatureDeploymentHistoryShadow] + return shadow || db.isDeploymentMetadataService() case WritingDeploymentHistoryOnly, WritingDeploymentHistoryShadow: return true default: @@ -513,24 +498,10 @@ func (db *DeploymentState) GetOrInitLineage() string { // If true, then Open reads the WAL and merges it in the state. If false, and WAL is present, Open returns an error. type WithRecovery bool -// Open loads deployment state and enables reads or writes according to mode: -// - ReadingFile: reads resources.json; state writes are disabled. -// - WritingFile: reads resources.json and opens a WAL for state writes. -// - WritingDeploymentHistoryOnly: reads and writes resources through DMS; no WAL is opened. -// - WritingDeploymentHistoryShadow: behaves like WritingFile and also records DMS history. -// History errors do not fail the deployment. -// - WritingDeploymentHistoryShadow: also persists a feature marker that rejects -// older clients that do not support history recording. -// -// All modes read deployment metadata and feature markers from resources.json. -// Open rejects modes that conflict with those markers. -// -// File modes recover a leftover WAL only when withRecovery is set. -// DMS modes reject any leftover WAL. -// -// DMS reads require a workspace client on ctx. DMS writes and history recording -// start after a version exists and InitializeOperationBuffer is called. -func (db *DeploymentState) Open(ctx context.Context, path string, mode Mode, withRecovery WithRecovery, dmsDeployment OpenDmsArgs) error { +// Open reads the file checkpoint and optionally recovers its WAL. It never enables writes. +// DMS-authoritative deployments must call LoadDMSState before planning. +// Call UpgradeToWrite after approval to enable writes. +func (db *DeploymentState) Open(ctx context.Context, path string, withRecovery WithRecovery) error { db.mu.Lock() defer db.mu.Unlock() @@ -538,7 +509,7 @@ func (db *DeploymentState) Open(ctx context.Context, path string, mode Mode, wit panic(fmt.Sprintf("state already opened: %v, cannot open %v", db.Path, path)) } - err := db.unlockedOpen(ctx, path, mode, withRecovery, dmsDeployment) + err := db.unlockedOpen(ctx, path, withRecovery) if err != nil { // A failed open must leave the receiver closed. unlockedOpen assigns // db.Path before every fallible step, so without this the receiver stays @@ -567,11 +538,10 @@ func (db *DeploymentState) reset() { } } -func (db *DeploymentState) unlockedOpen(ctx context.Context, path string, mode Mode, withRecovery WithRecovery, dmsDeployment OpenDmsArgs) error { +func (db *DeploymentState) unlockedOpen(ctx context.Context, path string, withRecovery WithRecovery) error { db.Path = path - db.mode = mode - db.DeploymentID = dmsDeployment.DeploymentID - db.VersionID = dmsDeployment.LastVersionID + db.mode = ReadingFile + db.DeploymentID, db.VersionID = "", 0 // The state file is the source of truth for whether this deployment records history: read it // first. @@ -607,7 +577,7 @@ func (db *DeploymentState) unlockedOpen(ctx context.Context, path string, mode M return fmt.Errorf("failed to stat WAL file %s: %w", walPath, err) default: // WAL exists switch { - case mode == WritingDeploymentHistoryOnly: + case db.isDeploymentMetadataService(): // A recorded deployment writes no WAL, so finding one means this state was written by a // deployment that did not record history. Refuse rather than discard it: the file is the // only record of that deploy's writes, and recording is refused below anyway. @@ -621,71 +591,6 @@ func (db *DeploymentState) unlockedOpen(ctx context.Context, path string, mode M } } - // Reconcile the config against the state. A brand-new deployment is where recording begins, so - // the config bootstraps the marker there. On an existing deployment the state is authoritative: - // the config can neither start recording one that was not (its resources would be created a - // second time) nor stop recording one that is (the service still holds it). - _, recorded := db.Data.Features[FeatureDeploymentHistory] - if mode == WritingDeploymentHistoryOnly && recorded && db.Data.Serial > dmsDeployment.LastVersionID { - return fmt.Errorf("inconsistent deployment state: resource state serial (%d) is ahead of deployment history version (%d)", db.Data.Serial, dmsDeployment.LastVersionID) - } - switch { - case mode == WritingDeploymentHistoryOnly && !recorded && len(db.Data.State) == 0: - if db.Data.Features == nil { - db.Data.Features = make(map[string]struct{}, 1) - } - db.Data.Features[FeatureDeploymentHistory] = struct{}{} - recorded = true - case mode == WritingDeploymentHistoryOnly && !recorded: - return errors.New(`enabling experimental.deployment_history for an existing deployment is not supported - -Run "databricks bundle destroy" first, then deploy again with deployment history enabled`) - case mode != WritingDeploymentHistoryOnly && recorded: - return ErrUnsettingRecording - } - - if recorded { - // The service is the source of truth for a recorded deployment; the file is a tombstone - // carrying only the marker, and applyDMSState loads the resources the service holds. - db.DeploymentID = dmsDeployment.DeploymentID - - if dmsDeployment.DeploymentID != "" { - w := cmdctx.WorkspaceClient(ctx) - recorded, err := retry.With(ctx, retry.IsTransientRead, func() ([]bundledeployments.Resource, error) { - return w.BundleDeployments.ListResourcesAll(ctx, bundledeployments.ListResourcesRequest{ - Parent: dms.DeploymentName(dmsDeployment.DeploymentID), - }) - }) - if err != nil { - return fmt.Errorf("listing resources from the deployment history service: %w", err) - } - resources := make([]dms.Resource, len(recorded)) - for i, res := range recorded { - // DMS keys drop the "resources." prefix; add it back for local state. - resources[i] = dms.Resource{ - Key: dms.StatePrefix + res.ResourceKey, - ID: res.ResourceId, - State: res.State, - } - } - if err := db.applyDMSState(resources); err != nil { - return err - } - } - - // The service owns the version number; the state file stores a copy. - // No deployment yet means no versions, hence zero. - db.VersionID = dmsDeployment.LastVersionID - } - - switch mode { - case ReadingFile: - return nil - case WritingDeploymentHistoryOnly: - return nil - case WritingFile, WritingDeploymentHistoryShadow: - return db.setupWal() - } return nil } @@ -955,17 +860,6 @@ func (db *DeploymentState) setupWal() error { CLIVersion: build.GetInfo().Version, Features: maps.Clone(db.Data.Features), } - switch db.mode { - case WritingDeploymentHistoryShadow: - if walHead.Features == nil { - walHead.Features = make(map[string]struct{}) - } - walHead.Features[FeatureDeploymentHistoryShadow] = struct{}{} - case WritingFile: - walHead.Features = map[string]struct{}{} - default: - // Other modes preserve the existing markers. - } if err := appendJSONLine(db.walFile, walHead); err != nil { // Remove the WAL created just above: O_EXCL means it is ours, and // without a header it carries nothing to recover. Leaving it makes @@ -1004,9 +898,7 @@ func (db *DeploymentState) AssertOpenedForReadOrWrite() { func (db *DeploymentState) AssertOpenedForRead() { db.AssertOpenedForReadOrWrite() - // DMS has no separate read-only mode. - // Planning reads DMS state while it is already in WritingDeploymentHistoryOnly. - if db.mode.isWritable() && db.mode != WritingDeploymentHistoryOnly { + if db.mode.isWritable() { panic("internal error: DeploymentState must be opened in read mode") } } diff --git a/bundle/direct/dstate/state_test.go b/bundle/direct/dstate/state_test.go index c5bfea2783b..033339aa4a6 100644 --- a/bundle/direct/dstate/state_test.go +++ b/bundle/direct/dstate/state_test.go @@ -25,14 +25,17 @@ func TestOpenSaveFinalizeRoundTrip(t *testing.T) { path := filepath.Join(t.TempDir(), "state.json") var db DeploymentState - require.NoError(t, db.Open(t.Context(), path, WritingFile, WithRecovery(true), OpenDmsArgs{})) + require.NoError(t, db.Open(t.Context(), path, WithRecovery(true))) + assert.NoFileExists(t, path+walSuffix) + assert.Panics(t, db.AssertOpenedForWrite) + require.NoError(t, db.UpgradeToWrite()) require.NoError(t, db.SaveState(t.Context(), "jobs.my_job", "123", map[string]string{"key": "val"}, nil)) mustFinalize(t, &db) // Re-open and verify persisted data. var db2 DeploymentState - require.NoError(t, db2.Open(t.Context(), path, ReadingFile, WithRecovery(false), OpenDmsArgs{})) + require.NoError(t, db2.Open(t.Context(), path, WithRecovery(false))) assert.Equal(t, 1, db2.Data.Serial) assert.Equal(t, "123", db2.GetResourceID("jobs.my_job")) mustFinalize(t, &db2) @@ -42,7 +45,7 @@ func TestFinalizeWithoutResourceWrites(t *testing.T) { for _, mode := range []Mode{WritingFile, WritingDeploymentHistoryShadow} { path := filepath.Join(t.TempDir(), "state.json") var db DeploymentState - require.NoError(t, db.Open(t.Context(), path, ReadingFile, WithRecovery(true), OpenDmsArgs{})) + require.NoError(t, db.Open(t.Context(), path, WithRecovery(true))) if mode == WritingDeploymentHistoryShadow { // A first plan already knows history is enabled, before a state file exists. db.Data.Features = map[string]struct{}{FeatureDeploymentHistoryShadow: {}} @@ -56,33 +59,36 @@ func TestFinalizeWithoutResourceWrites(t *testing.T) { continue } // Even an empty deployment must persist its history marker and stable lineage. - require.NoError(t, db.Open(t.Context(), path, ReadingFile, WithRecovery(false), OpenDmsArgs{})) + require.NoError(t, db.Open(t.Context(), path, WithRecovery(false))) assert.Equal(t, 1, db.Data.Serial) assert.NotEmpty(t, db.Data.Lineage) assert.Contains(t, db.StateFeatures(), FeatureDeploymentHistoryShadow) mustFinalize(t, &db) // Keeping history enabled without changes must not advance the file serial. - require.NoError(t, db.Open(t.Context(), path, WritingDeploymentHistoryShadow, WithRecovery(false), OpenDmsArgs{})) + require.NoError(t, db.Open(t.Context(), path, WithRecovery(false))) + require.NoError(t, db.UpgradeToWrite()) mustFinalize(t, &db) - require.NoError(t, db.Open(t.Context(), path, ReadingFile, WithRecovery(false), OpenDmsArgs{})) + require.NoError(t, db.Open(t.Context(), path, WithRecovery(false))) assert.Equal(t, 1, db.Data.Serial) mustFinalize(t, &db) // Turning history off also persists its marker change without resource operations. - require.NoError(t, db.Open(t.Context(), path, WritingFile, WithRecovery(false), OpenDmsArgs{})) + require.NoError(t, db.Open(t.Context(), path, WithRecovery(false))) + delete(db.Data.Features, FeatureDeploymentHistoryShadow) + require.NoError(t, db.UpgradeToWrite()) mustFinalize(t, &db) - require.NoError(t, db.Open(t.Context(), path, ReadingFile, WithRecovery(false), OpenDmsArgs{})) + require.NoError(t, db.Open(t.Context(), path, WithRecovery(false))) assert.Equal(t, 2, db.Data.Serial) assert.NotContains(t, db.StateFeatures(), FeatureDeploymentHistoryShadow) mustFinalize(t, &db) // Enabling history on existing state persists the marker, even without resource writes. - require.NoError(t, db.Open(t.Context(), path, ReadingFile, WithRecovery(false), OpenDmsArgs{})) + require.NoError(t, db.Open(t.Context(), path, WithRecovery(false))) db.Data.Features = map[string]struct{}{FeatureDeploymentHistoryShadow: {}} require.NoError(t, db.UpgradeToWrite()) mustFinalize(t, &db) - require.NoError(t, db.Open(t.Context(), path, ReadingFile, WithRecovery(false), OpenDmsArgs{})) + require.NoError(t, db.Open(t.Context(), path, WithRecovery(false))) assert.Equal(t, 3, db.Data.Serial) assert.Contains(t, db.StateFeatures(), FeatureDeploymentHistoryShadow) mustFinalize(t, &db) @@ -131,10 +137,12 @@ func TestDataForFileSerial(t *testing.T) { } } -func TestOpenUsesDMSVersionForSerial(t *testing.T) { +func TestLoadDMSStateUsesServiceVersionForSerial(t *testing.T) { server := testserver.New(t) server.Handle("GET", "/api/2.0/bundle/deployments/123/resources", func(req testserver.Request) any { - return bundledeployments.ListResourcesResponse{} + return bundledeployments.ListResourcesResponse{Resources: []bundledeployments.Resource{{ + ResourceKey: "jobs.foo", ResourceId: "123", State: `{"state":{"name":"foo"}}`, + }}} }) testserver.AddDefaultHandlers(server) client, err := databricks.NewWorkspaceClient(&databricks.Config{Host: server.URL, Token: "testtoken"}) @@ -223,16 +231,27 @@ func TestOpenUsesDMSVersionForSerial(t *testing.T) { ctx := cmdctx.SetWorkspaceClient(t.Context(), client) var db DeploymentState - err = db.Open(ctx, path, WritingDeploymentHistoryOnly, WithRecovery(false), OpenDmsArgs{ - DeploymentID: tt.deploymentID, LastVersionID: tt.lastVersionID, - }) + require.NoError(t, db.Open(ctx, path, WithRecovery(false))) + if !tt.unmarked { + err = db.LoadDMSState(ctx, tt.deploymentID, tt.lastVersionID) + } if tt.wantErr != "" { require.EqualError(t, err, "inconsistent deployment state: "+tt.wantErr) - assert.False(t, db.IsOpen()) + mustFinalize(t, &db) return } require.NoError(t, err) - assert.Equal(t, tt.lastVersionID, db.GetSerial()) + if tt.unmarked { + assert.Equal(t, tt.fileSerial, db.GetSerial()) + } else { + assert.Equal(t, tt.lastVersionID, db.GetSerial()) + if tt.deploymentID != "" { + assert.Equal(t, "123", db.GetResourceID("resources.jobs.foo")) + } + } + db.AssertOpenedForRead() + assert.Panics(t, db.AssertOpenedForWrite) + assert.NoFileExists(t, path+walSuffix) assert.Equal(t, "99.0.0", db.StateCLIVersion(), "preserve the last writer for downgrade warnings") mustFinalize(t, &db) }) @@ -288,10 +307,10 @@ func TestPanicOnDoubleOpen(t *testing.T) { path := filepath.Join(t.TempDir(), "state.json") var db DeploymentState - require.NoError(t, db.Open(t.Context(), path, WritingFile, WithRecovery(true), OpenDmsArgs{})) + require.NoError(t, db.Open(t.Context(), path, WithRecovery(true))) assert.Panics(t, func() { - _ = db.Open(t.Context(), path, WritingFile, WithRecovery(true), OpenDmsArgs{}) + _ = db.Open(t.Context(), path, WithRecovery(true)) }) mustFinalize(t, &db) } @@ -309,12 +328,13 @@ func TestCLIVersionRecordsLastWriter(t *testing.T) { require.NoError(t, os.WriteFile(path, []byte(seed), 0o600)) var db DeploymentState - require.NoError(t, db.Open(t.Context(), path, WritingFile, WithRecovery(true), OpenDmsArgs{})) + require.NoError(t, db.Open(t.Context(), path, WithRecovery(true))) + require.NoError(t, db.UpgradeToWrite()) require.NoError(t, db.SaveState(t.Context(), "resources.jobs.my_job", "123", map[string]string{"k": "v"}, nil)) mustFinalize(t, &db) var reopened DeploymentState - require.NoError(t, reopened.Open(t.Context(), path, ReadingFile, WithRecovery(false), OpenDmsArgs{})) + require.NoError(t, reopened.Open(t.Context(), path, WithRecovery(false))) assert.Equal(t, build.GetInfo().Version, reopened.Data.CLIVersion) assert.Equal(t, 2, reopened.Data.Serial) mustFinalize(t, &reopened) @@ -336,7 +356,7 @@ func TestHeaderOnlyWALDoesNotUpdateCLIVersion(t *testing.T) { require.NoError(t, os.WriteFile(walPath, append(headerLine, '\n'), 0o600)) var recovered DeploymentState - require.NoError(t, recovered.Open(t.Context(), path, ReadingFile, WithRecovery(true), OpenDmsArgs{})) + require.NoError(t, recovered.Open(t.Context(), path, WithRecovery(true))) assert.Equal(t, "0.1.2", recovered.Data.CLIVersion, "a header-only WAL wrote no state, so the version must not move") assert.Equal(t, 1, recovered.Data.Serial) mustFinalize(t, &recovered) @@ -348,12 +368,13 @@ func TestHeaderOnlyWALRecoveryDoesNotAdvanceSerial(t *testing.T) { // Commit serial 1 with one resource. var db DeploymentState - require.NoError(t, db.Open(t.Context(), path, WritingFile, WithRecovery(true), OpenDmsArgs{})) + require.NoError(t, db.Open(t.Context(), path, WithRecovery(true))) + require.NoError(t, db.UpgradeToWrite()) require.NoError(t, db.SaveState(t.Context(), "jobs.my_job", "123", map[string]string{}, nil)) mustFinalize(t, &db) var committed DeploymentState - require.NoError(t, committed.Open(t.Context(), path, ReadingFile, WithRecovery(false), OpenDmsArgs{})) + require.NoError(t, committed.Open(t.Context(), path, WithRecovery(false))) lineage := committed.Data.Lineage require.Equal(t, 1, committed.Data.Serial) mustFinalize(t, &committed) @@ -369,7 +390,7 @@ func TestHeaderOnlyWALRecoveryDoesNotAdvanceSerial(t *testing.T) { require.NoError(t, os.WriteFile(walPath, append(headerLine, '\n'), 0o600)) var recovered DeploymentState - require.NoError(t, recovered.Open(t.Context(), path, ReadingFile, WithRecovery(true), OpenDmsArgs{})) + require.NoError(t, recovered.Open(t.Context(), path, WithRecovery(true))) assert.Equal(t, 1, recovered.Data.Serial) assert.Equal(t, "123", recovered.GetResourceID("jobs.my_job")) assert.NoFileExists(t, walPath) @@ -398,17 +419,19 @@ func TestDeleteState(t *testing.T) { path := filepath.Join(t.TempDir(), "state.json") var db DeploymentState - require.NoError(t, db.Open(t.Context(), path, WritingFile, WithRecovery(true), OpenDmsArgs{})) + require.NoError(t, db.Open(t.Context(), path, WithRecovery(true))) + require.NoError(t, db.UpgradeToWrite()) require.NoError(t, db.SaveState(t.Context(), "jobs.my_job", "123", map[string]string{}, nil)) mustFinalize(t, &db) var db2 DeploymentState - require.NoError(t, db2.Open(t.Context(), path, WritingFile, WithRecovery(true), OpenDmsArgs{})) + require.NoError(t, db2.Open(t.Context(), path, WithRecovery(true))) + require.NoError(t, db2.UpgradeToWrite()) require.NoError(t, db2.DeleteState(t.Context(), "jobs.my_job", false)) mustFinalize(t, &db2) var db3 DeploymentState - require.NoError(t, db3.Open(t.Context(), path, ReadingFile, WithRecovery(false), OpenDmsArgs{})) + require.NoError(t, db3.Open(t.Context(), path, WithRecovery(false))) assert.Equal(t, 2, db3.Data.Serial) assert.Empty(t, db3.GetResourceID("jobs.my_job")) mustFinalize(t, &db3) @@ -420,7 +443,7 @@ func TestGetOrInitLineageReadableBeforeWriteAndPersisted(t *testing.T) { // Fresh state opened read-only, as the deploy does before planning: no // lineage yet. var db DeploymentState - require.NoError(t, db.Open(t.Context(), path, ReadingFile, WithRecovery(true), OpenDmsArgs{})) + require.NoError(t, db.Open(t.Context(), path, WithRecovery(true))) require.Empty(t, db.Data.Lineage) // GetOrInitLineage initializes the lineage and makes it readable before any @@ -437,7 +460,7 @@ func TestGetOrInitLineageReadableBeforeWriteAndPersisted(t *testing.T) { // Re-open: the persisted lineage matches the one read before the write. var reopened DeploymentState - require.NoError(t, reopened.Open(t.Context(), path, ReadingFile, WithRecovery(false), OpenDmsArgs{})) + require.NoError(t, reopened.Open(t.Context(), path, WithRecovery(false))) assert.Equal(t, lineage, reopened.Data.Lineage) mustFinalize(t, &reopened) } @@ -451,13 +474,13 @@ func TestOpenFailureLeavesStateClosed(t *testing.T) { require.NoError(t, os.WriteFile(path, []byte("{not json"), 0o600)) var db DeploymentState - require.Error(t, db.Open(t.Context(), path, WritingFile, WithRecovery(true), OpenDmsArgs{})) + require.Error(t, db.Open(t.Context(), path, WithRecovery(true))) assert.Empty(t, db.Path) // Once the state file is readable, the same receiver opens without panicking. seed := `{"state_version":2,"cli_version":"0.1.2","lineage":"test-lineage","serial":1,"state":{}}` require.NoError(t, os.WriteFile(path, []byte(seed), 0o600)) - require.NoError(t, db.Open(t.Context(), path, WritingFile, WithRecovery(true), OpenDmsArgs{})) + require.NoError(t, db.Open(t.Context(), path, WithRecovery(true))) assert.Equal(t, "test-lineage", db.Data.Lineage) mustFinalize(t, &db) } diff --git a/bundle/statemgmt/direct_migration.go b/bundle/statemgmt/direct_migration.go index 4fefd8504ac..c3e8aab97ac 100644 --- a/bundle/statemgmt/direct_migration.go +++ b/bundle/statemgmt/direct_migration.go @@ -194,9 +194,7 @@ func checkPlanOnTempState(ctx context.Context, b *bundle.Bundle, tempStatePath s var planBundle direct.DeploymentBundle - // This plan is not created with the deployment history feature enabled, - // so we can safely pass false for withDeploymentHistory. - if err := planBundle.StateDB.Open(planCtx, tempStatePath, dstate.ReadingFile, dstate.WithRecovery(false), dstate.OpenDmsArgs{}); err != nil { + if err := planBundle.StateDB.Open(planCtx, tempStatePath, dstate.WithRecovery(false)); err != nil { return nil, fmt.Errorf("opening migrated state for plan check: %w", err) } diff --git a/cmd/bundle/generate/state.go b/cmd/bundle/generate/state.go index 2868be7166d..2cb5051e64f 100644 --- a/cmd/bundle/generate/state.go +++ b/cmd/bundle/generate/state.go @@ -37,7 +37,7 @@ func loadStateForGenerate(ctx context.Context, b *bundle.Bundle) context.Context var state statemgmt.ExportedResourcesMap if stateDesc.Engine.IsDirect() { - if err := utils.OpenDirectStateForRead(ctx, b, stateDesc); err != nil { + if err := utils.OpenDirectStateForRead(ctx, b); err != nil { logdiag.LogError(ctx, err) return ctx } diff --git a/cmd/bundle/utils/process.go b/cmd/bundle/utils/process.go index 68798a4bb01..5e02abcc37b 100644 --- a/cmd/bundle/utils/process.go +++ b/cmd/bundle/utils/process.go @@ -366,20 +366,17 @@ func ProcessBundleRet(cmd *cobra.Command, opts ProcessOptions) (b *bundle.Bundle // Open direct engine state once for all subsequent operations (ExportState, CalculatePlan, Apply, etc.) // A migrated-from-Terraform state is already open (seeded in memory above), so skip the disk open. needDirectState := stateDesc.Engine.IsDirect() && needsState - var localPath string if needDirectState && !b.DeploymentBundle.StateDB.IsOpen() { - _, localPath = b.StateFilenameDirect(ctx) - if !stateDesc.IsDMS() { - if err := b.DeploymentBundle.StateDB.Open(ctx, localPath, dstate.ReadingFile, dstate.WithRecovery(true), dstate.OpenDmsArgs{}); err != nil { - logdiag.LogError(ctx, err) - return b, stateDesc, root.ErrAlreadyPrinted - } + _, localPath := b.StateFilenameDirect(ctx) + if err := b.DeploymentBundle.StateDB.Open(ctx, localPath, dstate.WithRecovery(true)); err != nil { + logdiag.LogError(ctx, err) + return b, stateDesc, root.ErrAlreadyPrinted } } stateMode := dstate.WritingFile switch { - case stateDesc.IsDMS(): + case b.DeploymentBundle.StateDB.IsDeploymentMetadataService(): // DMS-only state must keep its service-backed source. stateMode = dstate.WritingDeploymentHistoryOnly case opts.Deploy || opts.PreDeployChecks: @@ -416,16 +413,23 @@ func ProcessBundleRet(cmd *cobra.Command, opts ProcessOptions) (b *bundle.Bundle logdiag.LogError(ctx, err) return b, stateDesc, root.ErrAlreadyPrinted } - - // Stamp the deployment and the version this run records onto every job and pipeline so - // the plan carries them. version_id is always known (last recorded + 1); deployment_id - // does not exist until a first deploy creates it, so it is left off here and the deploy - // phase stamps the created id. lastVersionID, err := parseLastVersionID(dmsDeployment) if err != nil { logdiag.LogError(ctx, err) return b, stateDesc, root.ErrAlreadyPrinted } + if !cmdctx.HasWorkspaceClient(ctx) { + ctx = cmdctx.SetWorkspaceClient(ctx, b.WorkspaceClient(ctx)) + } + if err := b.DeploymentBundle.StateDB.LoadDMSState(ctx, dmsDeploymentID, lastVersionID); err != nil { + logdiag.LogError(ctx, err) + return b, stateDesc, root.ErrAlreadyPrinted + } + + // Stamp the deployment and the version this run records onto every job and pipeline so + // the plan carries them. version_id is always known (last recorded + 1); deployment_id + // does not exist until a first deploy creates it, so it is left off here and the deploy + // phase stamps the created id. nextVersion := lastVersionID + 1 muts := []bundle.Mutator{metadata.AnnotateDeploymentVersion(nextVersion)} if dmsDeploymentID != "" { @@ -439,14 +443,6 @@ func ProcessBundleRet(cmd *cobra.Command, opts ProcessOptions) (b *bundle.Bundle if logdiag.HasError(ctx) { return b, stateDesc, root.ErrAlreadyPrinted } - // StateDB.Open builds the DMS client from the workspace client on the context. - if !cmdctx.HasWorkspaceClient(ctx) { - ctx = cmdctx.SetWorkspaceClient(ctx, b.WorkspaceClient(ctx)) - } - if err := b.DeploymentBundle.StateDB.Open(ctx, localPath, dstate.WritingDeploymentHistoryOnly, dstate.WithRecovery(false), dstate.OpenDmsArgs{DeploymentID: dmsDeploymentID, LastVersionID: lastVersionID}); err != nil { - logdiag.LogError(ctx, err) - return b, stateDesc, root.ErrAlreadyPrinted - } } if needDirectState { @@ -684,32 +680,25 @@ func parseLastVersionID(dmsDeployment *bundledeployments.Deployment) (int, error return v, nil } -// OpenDirectStateForRead opens the direct-engine state database read-only. When the bundle -// records deployment history the local state file is only a tombstone, so the resources are -// read from the deployment metadata service instead. -func OpenDirectStateForRead(ctx context.Context, b *bundle.Bundle, stateDesc *statemgmt.StateDesc) error { +// OpenDirectStateForRead opens the file checkpoint and loads DMS resources when required. +func OpenDirectStateForRead(ctx context.Context, b *bundle.Bundle) error { _, localPath := b.StateFilenameDirect(ctx) - if !stateDesc.IsDMS() { - if err := b.DeploymentBundle.StateDB.Open(ctx, localPath, dstate.ReadingFile, dstate.WithRecovery(true), dstate.OpenDmsArgs{}); err != nil { - return err - } - return nil - } - - dmsDeploymentID, dmsDeployment, err := fetchDeploymentFromStatePath(ctx, b.WorkspaceClient(ctx), b.Config.Workspace.StatePath) - if err != nil { + if err := b.DeploymentBundle.StateDB.Open(ctx, localPath, dstate.WithRecovery(true)); err != nil { return err } - lastVersionID, err := parseLastVersionID(dmsDeployment) - if err != nil { - return err - } - // StateDB.Open builds the DMS client from the workspace client on the context, so ensure one is set. - if !cmdctx.HasWorkspaceClient(ctx) { - ctx = cmdctx.SetWorkspaceClient(ctx, b.WorkspaceClient(ctx)) - } - if err := b.DeploymentBundle.StateDB.Open(ctx, localPath, dstate.WritingDeploymentHistoryOnly, dstate.WithRecovery(false), dstate.OpenDmsArgs{DeploymentID: dmsDeploymentID, LastVersionID: lastVersionID}); err != nil { - return err + if b.DeploymentBundle.StateDB.IsDeploymentMetadataService() { + deploymentID, deployment, err := fetchDeploymentFromStatePath(ctx, b.WorkspaceClient(ctx), b.Config.Workspace.StatePath) + if err != nil { + return err + } + lastVersionID, err := parseLastVersionID(deployment) + if err != nil { + return err + } + if !cmdctx.HasWorkspaceClient(ctx) { + ctx = cmdctx.SetWorkspaceClient(ctx, b.WorkspaceClient(ctx)) + } + return b.DeploymentBundle.StateDB.LoadDMSState(ctx, deploymentID, lastVersionID) } return nil }