PSS: Allow pluggable state store configuration to be stored in a plan file (#37956)

* refactor: Rename Meta's backendState field to backendConfigState

This helps with navigating ambiguity around the word backend. The new name should indicate that the value represents a `backend` block, not a more general interpretation of what a backend is.

* fix: Only set backendConfigState to synthetic object if it's nil due to a lack of data. Don't change it if pluggable state storage is in use.

* feat: Enable recording a state store's details in an Operation, and using that data when creating a plan file.

* fix: Include provider config when writing a plan file using pluggable state storage

* fix: Having `backendConfigState` be nil may be valid, but it definitely isn't valid for `stateStoreConfigState` to be nil

When backendConfigState is nil it means that an implied local backend is in use, i.e. there is no backend block in the config.

* test: Add integration test showing that a plan command creates a plan file with the expected state_store configuration data

* refactor: Apply suggestion from @radeksimko

Co-authored-by: Radek Simko <radeksimko@users.noreply.github.com>

* fix: Allow panics to occur if an unexpected implementation of `backend.Backend` is encountered when managing a state store

* docs: Add code comment explaining the current situation with passing backend config state to downstream logic.

In future this should be simplified, either via refactoring or changes affecting the implied local backend

---------

Co-authored-by: Radek Simko <radeksimko@users.noreply.github.com>

internal/backend/backendrun/operation.go

@@ -73,14 +73,20 @@ type Operation struct {
 
 	// PlanId is an opaque value that backends can use to execute a specific
 	// plan for an apply operation.
-	//
+	PlanId      string
+	PlanRefresh bool   // PlanRefresh will do a refresh before a plan
+	PlanOutPath string // PlanOutPath is the path to save the plan
+
 	// PlanOutBackend is the backend to store with the plan. This is the
 	// backend that will be used when applying the plan.
-	PlanId         string
-	PlanRefresh    bool   // PlanRefresh will do a refresh before a plan
-	PlanOutPath    string // PlanOutPath is the path to save the plan
+	// Only one of PlanOutBackend or PlanOutStateStore may be set.
 	PlanOutBackend *plans.Backend
 
+	// PlanOutStateStore is the state_store to store with the plan. This is the
+	// state store that will be used when applying the plan.
+	// Only one of PlanOutBackend or PlanOutStateStore may be set
+	PlanOutStateStore *plans.StateStore
+
 	// ConfigDir is the path to the directory containing the configuration's
 	// root module.
 	ConfigDir string

internal/backend/local/backend_plan.go

@@ -149,16 +149,22 @@ func (b *Local) opPlan(
 
 	// Save the plan to disk
 	if path := op.PlanOutPath; path != "" {
-		if op.PlanOutBackend == nil {
+		switch {
+		case op.PlanOutStateStore != nil:
+			plan.StateStore = op.PlanOutStateStore
+		case op.PlanOutBackend != nil:
+			plan.Backend = op.PlanOutBackend
+		default:
 			// This is always a bug in the operation caller; it's not valid
-			// to set PlanOutPath without also setting PlanOutBackend.
+			// to set PlanOutPath without also setting PlanOutStateStore or PlanOutBackend.
+			// Even when there is no state_store or backend block in the configuration, there should be a PlanOutBackend
+			// describing the implied local backend.
 			diags = diags.Append(fmt.Errorf(
-				"PlanOutPath set without also setting PlanOutBackend (this is a bug in Terraform)"),
+				"PlanOutPath set without also setting PlanOutStateStore or PlanOutBackend (this is a bug in Terraform)"),
 			)
 			op.ReportResult(runningOp, diags)
 			return
 		}
-		plan.Backend = op.PlanOutBackend
 
 		// We may have updated the state in the refresh step above, but we
 		// will freeze that updated state in the plan file for now and

internal/backend/local/backend_plan_test.go

@@ -12,6 +12,7 @@ import (
 
 	"github.com/zclconf/go-cty/cty"
 
+	"github.com/hashicorp/go-version"
 	"github.com/hashicorp/terraform/internal/addrs"
 	"github.com/hashicorp/terraform/internal/backend/backendrun"
 	"github.com/hashicorp/terraform/internal/command/arguments"
@@ -913,3 +914,155 @@ func TestLocal_invalidOptions(t *testing.T) {
 		t.Fatal("expected error output")
 	}
 }
+
+// Checks if the state store info set on an Operation makes it into the resulting Plan
+func TestLocal_plan_withStateStore(t *testing.T) {
+	b := TestLocal(t)
+
+	// Note: the mock provider doesn't include an implementation of
+	// pluggable state storage, but that's not needed for this test.
+	TestLocalProvider(t, b, "test", planFixtureSchema())
+	mockAddr := addrs.NewDefaultProvider("test")
+	providerVersion := version.Must(version.NewSemver("0.0.1"))
+	storeType := "test_foobar"
+	defaultWorkspace := "default"
+
+	testStateFile(t, b.StatePath, testPlanState_withDataSource())
+
+	outDir := t.TempDir()
+	planPath := filepath.Join(outDir, "plan.tfplan")
+
+	// Note: the config doesn't include a state_store block. Instead,
+	// that data is provided below when assigning a value to op.PlanOutStateStore.
+	// Usually that data is set as a result of parsing configuration.
+	op, configCleanup, _ := testOperationPlan(t, "./testdata/plan")
+	defer configCleanup()
+	op.PlanMode = plans.NormalMode
+	op.PlanRefresh = true
+	op.PlanOutPath = planPath
+	storeCfg := cty.ObjectVal(map[string]cty.Value{
+		"path": cty.StringVal(b.StatePath),
+	})
+	storeCfgRaw, err := plans.NewDynamicValue(storeCfg, storeCfg.Type())
+	if err != nil {
+		t.Fatal(err)
+	}
+	providerCfg := cty.ObjectVal(map[string]cty.Value{}) // Empty as the mock provider has no schema for the provider
+	providerCfgRaw, err := plans.NewDynamicValue(providerCfg, providerCfg.Type())
+	if err != nil {
+		t.Fatal(err)
+	}
+	op.PlanOutStateStore = &plans.StateStore{
+		Type:   storeType,
+		Config: storeCfgRaw,
+		Provider: &plans.Provider{
+			Source:  &mockAddr,
+			Version: providerVersion,
+			Config:  providerCfgRaw,
+		},
+		Workspace: defaultWorkspace,
+	}
+
+	run, err := b.Operation(context.Background(), op)
+	if err != nil {
+		t.Fatalf("bad: %s", err)
+	}
+	<-run.Done()
+	if run.Result != backendrun.OperationSuccess {
+		t.Fatalf("plan operation failed")
+	}
+
+	if run.PlanEmpty {
+		t.Fatal("plan should not be empty")
+	}
+
+	plan := testReadPlan(t, planPath)
+
+	// The plan should contain details about the state store
+	if plan.StateStore == nil {
+		t.Fatalf("Expected plan to describe a state store, but data was missing")
+	}
+	// The plan should NOT contain details about a backend
+	if plan.Backend != nil {
+		t.Errorf("Expected plan to not describe a backend because a state store is in use, but data was present:\n plan.Backend = %v", plan.Backend)
+	}
+
+	if plan.StateStore.Type != storeType {
+		t.Errorf("Expected plan to describe a state store with type %s, but got %s", storeType, plan.StateStore.Type)
+	}
+	if plan.StateStore.Workspace != defaultWorkspace {
+		t.Errorf("Expected plan to describe a state store with workspace %s, but got %s", defaultWorkspace, plan.StateStore.Workspace)
+	}
+	if !plan.StateStore.Provider.Source.Equals(mockAddr) {
+		t.Errorf("Expected plan to describe a state store with provider address %s, but got %s", mockAddr, plan.StateStore.Provider.Source)
+	}
+	if !plan.StateStore.Provider.Version.Equal(providerVersion) {
+		t.Errorf("Expected plan to describe a state store with provider version %s, but got %s", providerVersion, plan.StateStore.Provider.Version)
+	}
+}
+
+// Checks if the backend info set on an Operation makes it into the resulting Plan
+func TestLocal_plan_withBackend(t *testing.T) {
+	b := TestLocal(t)
+
+	TestLocalProvider(t, b, "test", planFixtureSchema())
+
+	testStateFile(t, b.StatePath, testPlanState_withDataSource())
+
+	outDir := t.TempDir()
+	planPath := filepath.Join(outDir, "plan.tfplan")
+
+	// Note: the config doesn't include a backend block. Instead,
+	// that data is provided below when assigning a value to op.PlanOutBackend.
+	// Usually that data is set as a result of parsing configuration.
+	op, configCleanup, _ := testOperationPlan(t, "./testdata/plan")
+	defer configCleanup()
+	op.PlanMode = plans.NormalMode
+	op.PlanRefresh = true
+	op.PlanOutPath = planPath
+	cfg := cty.ObjectVal(map[string]cty.Value{
+		"path": cty.StringVal(b.StatePath),
+	})
+	cfgRaw, err := plans.NewDynamicValue(cfg, cfg.Type())
+	if err != nil {
+		t.Fatal(err)
+	}
+	backendType := "foobar"
+	defaultWorkspace := "default"
+	op.PlanOutBackend = &plans.Backend{
+		Type:      backendType,
+		Config:    cfgRaw,
+		Workspace: defaultWorkspace,
+	}
+
+	run, err := b.Operation(context.Background(), op)
+	if err != nil {
+		t.Fatalf("bad: %s", err)
+	}
+	<-run.Done()
+	if run.Result != backendrun.OperationSuccess {
+		t.Fatalf("plan operation failed")
+	}
+
+	if run.PlanEmpty {
+		t.Fatal("plan should not be empty")
+	}
+
+	plan := testReadPlan(t, planPath)
+
+	// The plan should contain details about the backend
+	if plan.Backend == nil {
+		t.Fatalf("Expected plan to describe a backend, but data was missing")
+	}
+	// The plan should NOT contain details about a state store
+	if plan.StateStore != nil {
+		t.Errorf("Expected plan to not describe a state store because a backend is in use, but data was present:\n plan.StateStore = %v", plan.StateStore)
+	}
+
+	if plan.Backend.Type != backendType {
+		t.Errorf("Expected plan to describe a backend with type %s, but got %s", backendType, plan.Backend.Type)
+	}
+	if plan.Backend.Workspace != defaultWorkspace {
+		t.Errorf("Expected plan to describe a backend with workspace %s, but got %s", defaultWorkspace, plan.Backend.Workspace)
+	}
+}

internal/command/meta.go

@@ -208,8 +208,12 @@ type Meta struct {
 	// It is initialized on first use.
 	configLoader *configload.Loader
 
-	// backendState is the currently active backend state
-	backendState *workdir.BackendConfigState
+	// backendConfigState is the currently active backend state.
+	// This is used when creating plan files.
+	backendConfigState *workdir.BackendConfigState
+	// stateStoreConfigState is the currently active state_store state.
+	// This is used when creating plan files.
+	stateStoreConfigState *workdir.StateStoreConfigState
 
 	// Variables for the context (private)
 	variableArgs arguments.FlagNameValueSlice

internal/command/meta_backend.go

@@ -29,6 +29,7 @@ import (
 	"github.com/hashicorp/terraform/internal/backend"
 	"github.com/hashicorp/terraform/internal/backend/backendrun"
 	backendInit "github.com/hashicorp/terraform/internal/backend/init"
+	"github.com/hashicorp/terraform/internal/backend/local"
 	backendLocal "github.com/hashicorp/terraform/internal/backend/local"
 	backendPluggable "github.com/hashicorp/terraform/internal/backend/pluggable"
 	"github.com/hashicorp/terraform/internal/cloud"
@@ -37,6 +38,7 @@ import (
 	"github.com/hashicorp/terraform/internal/command/views"
 	"github.com/hashicorp/terraform/internal/command/workdir"
 	"github.com/hashicorp/terraform/internal/configs"
+	"github.com/hashicorp/terraform/internal/configs/configschema"
 	"github.com/hashicorp/terraform/internal/depsfile"
 	"github.com/hashicorp/terraform/internal/didyoumean"
 	"github.com/hashicorp/terraform/internal/getproviders/providerreqs"
@@ -221,13 +223,14 @@ func (m *Meta) Backend(opts *BackendOpts) (backendrun.OperationsBackend, tfdiags
 	// the user, since the local backend should only be used when learning or
 	// in exceptional cases and so it's better to help the user learn that
 	// by introducing it as a concept.
-	if m.backendState == nil {
+	stateStoreInUse := opts.StateStoreConfig != nil
+	if !stateStoreInUse && m.backendConfigState == nil {
 		// NOTE: This synthetic object is intentionally _not_ retained in the
 		// on-disk record of the backend configuration, which was already dealt
 		// with inside backendFromConfig, because we still need that codepath
 		// to be able to recognize the lack of a config as distinct from
 		// explicitly setting local until we do some more refactoring here.
-		m.backendState = &workdir.BackendConfigState{
+		m.backendConfigState = &workdir.BackendConfigState{
 			Type:      "local",
 			ConfigRaw: json.RawMessage("{}"),
 		}
@@ -440,13 +443,38 @@ func (m *Meta) Operation(b backend.Backend, vt arguments.ViewType) *backendrun.O
 		// here first is a bug, so panic.
 		panic(fmt.Sprintf("invalid workspace: %s", err))
 	}
-	planOutBackend, err := m.backendState.PlanData(schema, nil, workspace)
-	if err != nil {
-		// Always indicates an implementation error in practice, because
-		// errors here indicate invalid encoding of the backend configuration
-		// in memory, and we should always have validated that by the time
-		// we get here.
-		panic(fmt.Sprintf("failed to encode backend configuration for plan: %s", err))
+
+	var planOutBackend *plans.Backend
+	var planOutStateStore *plans.StateStore
+	switch {
+	case m.backendConfigState != nil && m.stateStoreConfigState != nil:
+		// Both set
+		panic("failed to encode backend configuration for plan: both backend and state_store data present but they are mutually exclusive")
+	case m.stateStoreConfigState != nil:
+		// To access the provider schema, we need to access the underlying backends
+		var providerSchema *configschema.Block
+		lb := b.(*local.Local)
+		p := lb.Backend.(*backendPluggable.Pluggable)
+		providerSchema = p.ProviderSchema()
+
+		planOutStateStore, err = m.stateStoreConfigState.PlanData(schema, providerSchema, workspace)
+		if err != nil {
+			// Always indicates an implementation error in practice, because
+			// errors here indicate invalid encoding of the state_store configuration
+			// in memory, and we should always have validated that by the time
+			// we get here.
+			panic(fmt.Sprintf("failed to encode state_store configuration for plan: %s", err))
+		}
+	default:
+		// Either backendConfigState is set, or it's nil; PlanData method can handle either.
+		planOutBackend, err = m.backendConfigState.PlanData(schema, nil, workspace)
+		if err != nil {
+			// Always indicates an implementation error in practice, because
+			// errors here indicate invalid encoding of the backend configuration
+			// in memory, and we should always have validated that by the time
+			// we get here.
+			panic(fmt.Sprintf("failed to encode backend configuration for plan: %s", err))
+		}
 	}
 
 	stateLocker := clistate.NewNoopLocker()
@@ -465,8 +493,11 @@ func (m *Meta) Operation(b backend.Backend, vt arguments.ViewType) *backendrun.O
 		log.Printf("[WARN] Failed to load dependency locks while preparing backend operation (ignored): %s", diags.Err().Error())
 	}
 
-	return &backendrun.Operation{
-		PlanOutBackend:  planOutBackend,
+	op := &backendrun.Operation{
+		// These two fields are mutually exclusive; one is being assigned a nil value below.
+		PlanOutBackend:    planOutBackend,
+		PlanOutStateStore: planOutStateStore,
+
 		Targets:         m.targets,
 		UIIn:            m.UIInput(),
 		UIOut:           m.Ui,
@@ -474,6 +505,12 @@ func (m *Meta) Operation(b backend.Backend, vt arguments.ViewType) *backendrun.O
 		StateLocker:     stateLocker,
 		DependencyLocks: depLocks,
 	}
+
+	if op.PlanOutBackend != nil && op.PlanOutStateStore != nil {
+		panic("failed to prepare operation: both backend and state_store configurations are present")
+	}
+
+	return op
 }
 
 // backendConfig returns the local configuration for the backend
@@ -727,10 +764,28 @@ func (m *Meta) backendFromConfig(opts *BackendOpts) (backend.Backend, tfdiags.Di
 
 	// Upon return, we want to set the state we're using in-memory so that
 	// we can access it for commands.
-	m.backendState = nil
+	//
+	// Currently the only command using these values is the `plan` command,
+	// which records the data in the plan file.
+	m.backendConfigState = nil
+	m.stateStoreConfigState = nil
 	defer func() {
-		if s := sMgr.State(); s != nil && !s.Backend.Empty() {
-			m.backendState = s.Backend
+		s := sMgr.State()
+		switch {
+		case s == nil:
+			// Do nothing
+			/* If there is no backend state file then either:
+			1. The working directory isn't initialized yet.
+				The user is either in the process of running an init command, in which case the values set via this deferred function will not be used,
+				or they are performing a non-init command that will be interrupted by an error before these values are used in downstream
+			2. There isn't any backend or state_store configuration and an implied local backend is in use.
+				This is valid and will mean m.backendConfigState is nil until the calling code adds a synthetic object in:
+				https://github.com/hashicorp/terraform/blob/3eea12a1d810a17e9c8e43cf7774817641ca9bc1/internal/command/meta_backend.go#L213-L234
+			*/
+		case !s.Backend.Empty():
+			m.backendConfigState = s.Backend
+		case !s.StateStore.Empty():
+			m.stateStoreConfigState = s.StateStore
 		}
 	}()
 
@@ -1781,11 +1836,12 @@ func (m *Meta) stateStore_C_s(c *configs.StateStore, stateStoreHash int, backend
 		},
 	}
 	s.StateStore.SetConfig(storeConfigVal, b.ConfigSchema())
-	if plug, ok := b.(*backendPluggable.Pluggable); ok {
-		// We need to convert away from backend.Backend interface to use the method
-		// for accessing the provider schema.
-		s.StateStore.Provider.SetConfig(providerConfigVal, plug.ProviderSchema())
-	}
+
+	// We need to briefly convert away from backend.Backend interface to use the method
+	// for accessing the provider schema. In this method we _always_ expect the concrete value
+	// to be backendPluggable.Pluggable.
+	plug := b.(*backendPluggable.Pluggable)
+	s.StateStore.Provider.SetConfig(providerConfigVal, plug.ProviderSchema())
 
 	// Verify that selected workspace exists in the state store.
 	if opts.Init && b != nil {

internal/command/plan_test.go

@@ -22,11 +22,13 @@ import (
 	"github.com/hashicorp/terraform/internal/addrs"
 	backendinit "github.com/hashicorp/terraform/internal/backend/init"
 	"github.com/hashicorp/terraform/internal/checks"
+	"github.com/hashicorp/terraform/internal/command/clistate"
 	"github.com/hashicorp/terraform/internal/configs/configschema"
 	"github.com/hashicorp/terraform/internal/plans"
 	"github.com/hashicorp/terraform/internal/providers"
 	testing_provider "github.com/hashicorp/terraform/internal/providers/testing"
 	"github.com/hashicorp/terraform/internal/states"
+	"github.com/hashicorp/terraform/internal/states/statefile"
 	"github.com/hashicorp/terraform/internal/tfdiags"
 )
 
@@ -537,7 +539,6 @@ func TestPlan_outBackend_withWorkspace(t *testing.T) {
 	}
 
 	plan := testReadPlan(t, outPath)
-
 	if got, want := plan.Backend.Type, expectedBackendType; got != want {
 		t.Errorf("wrong backend type %q; want %q", got, want)
 	}
@@ -546,6 +547,156 @@ func TestPlan_outBackend_withWorkspace(t *testing.T) {
 	}
 }
 
+// When using "-out" with a state store, the plan should encode the state store config
+func TestPlan_outStateStore(t *testing.T) {
+	// Create a temporary working directory with state_store config
+	td := t.TempDir()
+	testCopyDir(t, testFixturePath("plan-out-state-store"), td)
+	t.Chdir(td)
+
+	// Make state that resembles the resource defined in the test fixture
+	originalState := states.BuildState(func(s *states.SyncState) {
+		s.SetResourceInstanceCurrent(
+			addrs.Resource{
+				Mode: addrs.ManagedResourceMode,
+				Type: "test_instance",
+				Name: "foo",
+			}.Instance(addrs.NoKey).Absolute(addrs.RootModuleInstance),
+			&states.ResourceInstanceObjectSrc{
+				AttrsJSON: []byte(`{"id":"bar","ami":"bar"}`),
+				Status:    states.ObjectReady,
+			},
+			addrs.AbsProviderConfig{
+				Provider: addrs.NewDefaultProvider("test"),
+				Module:   addrs.RootModule,
+			},
+		)
+	})
+	var stateBuf bytes.Buffer
+	if err := statefile.Write(statefile.New(originalState, "", 1), &stateBuf); err != nil {
+		t.Fatalf("error during test setup: %s", err)
+	}
+	stateBytes := stateBuf.Bytes()
+
+	// Make a mock provider that:
+	// 1) will return the state defined above.
+	// 2) has a schema for the resource being managed in this test.
+	mock := mockPluggableStateStorageProvider()
+	mock.MockStates = map[string]interface{}{
+		"default": stateBytes,
+	}
+	mock.GetProviderSchemaResponse.ResourceTypes = map[string]providers.Schema{
+		"test_instance": {
+			Body: &configschema.Block{
+				Attributes: map[string]*configschema.Attribute{
+					"id": {
+						Type:     cty.String,
+						Computed: true,
+					},
+					"ami": {
+						Type:     cty.String,
+						Optional: true,
+					},
+				},
+			},
+		},
+	}
+	mock.PlanResourceChangeFn = func(req providers.PlanResourceChangeRequest) providers.PlanResourceChangeResponse {
+		return providers.PlanResourceChangeResponse{
+			PlannedState: req.ProposedNewState,
+		}
+	}
+	view, done := testView(t)
+	c := &PlanCommand{
+		Meta: Meta{
+			AllowExperimentalFeatures: true,
+			testingOverrides:          metaOverridesForProvider(mock),
+			View:                      view,
+		},
+	}
+
+	outPath := "foo"
+	args := []string{
+		"-out", outPath,
+		"-no-color",
+	}
+	code := c.Run(args)
+	output := done(t)
+	if code != 0 {
+		t.Logf("stdout: %s", output.Stdout())
+		t.Fatalf("plan command failed with exit code %d\n\n%s", code, output.Stderr())
+	}
+
+	plan := testReadPlan(t, outPath)
+	if !plan.Changes.Empty() {
+		t.Fatalf("Expected empty plan to be written to plan file, got: %s", spew.Sdump(plan))
+	}
+
+	if plan.Backend != nil {
+		t.Errorf("expected the plan file to not describe a backend, but got %#v", plan.Backend)
+	}
+	if plan.StateStore == nil {
+		t.Errorf("expected the plan file to describe a state store, but it's empty: %#v", plan.StateStore)
+	}
+	if got, want := plan.StateStore.Workspace, "default"; got != want {
+		t.Errorf("wrong workspace %q; want %q", got, want)
+	}
+	{
+		// Comparing the plan's description of the state store
+		// to the backend state file's description of the state store:
+		statePath := ".terraform/terraform.tfstate"
+		sMgr := &clistate.LocalState{Path: statePath}
+		if err := sMgr.RefreshState(); err != nil {
+			t.Fatal(err)
+		}
+		s := sMgr.State() // The plan should resemble this.
+
+		if !plan.StateStore.Provider.Version.Equal(s.StateStore.Provider.Version) {
+			t.Fatalf("wrong provider version, got %q; want %q",
+				plan.StateStore.Provider.Version,
+				s.StateStore.Provider.Version,
+			)
+		}
+		if !plan.StateStore.Provider.Source.Equals(*s.StateStore.Provider.Source) {
+			t.Fatalf("wrong provider source, got %q; want %q",
+				plan.StateStore.Provider.Source,
+				s.StateStore.Provider.Source,
+			)
+		}
+
+		// Is the provider config data correct?
+		providerSchema := mock.GetProviderSchemaResponse.Provider
+		providerTy := providerSchema.Body.ImpliedType()
+		pGot, err := plan.StateStore.Provider.Config.Decode(providerTy)
+		if err != nil {
+			t.Fatalf("failed to decode provider config in plan: %s", err)
+		}
+		pWant, err := s.StateStore.Provider.Config(providerSchema.Body)
+		if err != nil {
+			t.Fatalf("failed to decode cached provider config: %s", err)
+		}
+		if !pWant.RawEquals(pGot) {
+			t.Errorf("wrong provider config\ngot:  %#v\nwant: %#v", pGot, pWant)
+		}
+
+		// Is the store config data correct?
+		storeSchema := mock.GetProviderSchemaResponse.StateStores["test_store"]
+		ty := storeSchema.Body.ImpliedType()
+		sGot, err := plan.StateStore.Config.Decode(ty)
+		if err != nil {
+			t.Fatalf("failed to decode state store config in plan: %s", err)
+		}
+
+		sWant, err := s.StateStore.Config(storeSchema.Body)
+		if err != nil {
+			t.Fatalf("failed to decode cached state store config: %s", err)
+		}
+		if !sWant.RawEquals(sGot) {
+			t.Errorf("wrong state store config\ngot:  %#v\nwant: %#v", sGot, sWant)
+		}
+	}
+}
+
 func TestPlan_refreshFalse(t *testing.T) {
 	// Create a temporary working directory that is empty
 	td := t.TempDir()

internal/command/testdata/plan-out-state-store/.terraform.lock.hcl

@@ -0,0 +1,6 @@
+# This file is maintained automatically by "terraform init".
+# Manual edits may be lost in future updates.
+
+provider "registry.terraform.io/hashicorp/test" {
+  version = "1.2.3"
+}

internal/command/testdata/plan-out-state-store/.terraform/terraform.tfstate

@@ -0,0 +1,19 @@
+{
+    "version": 3,
+    "serial": 0,
+    "lineage": "666f9301-7e65-4b19-ae23-71184bb19b03",
+    "state_store": {
+        "type": "test_store",
+        "config": {
+            "value": "foobar"
+        },
+        "provider": {
+            "version": "1.2.3",
+            "source": "registry.terraform.io/hashicorp/test",
+            "config": {
+                "region": null
+            }
+        },
+        "hash": 4158988729
+    }
+}

internal/command/testdata/plan-out-state-store/main.tf

@@ -0,0 +1,17 @@
+terraform {
+  required_providers {
+    test = {
+      source  = "hashicorp/test"
+      version = "1.2.3"
+    }
+  }
+  state_store "test_store" {
+    provider "test" {}
+
+    value = "foobar"
+  }
+}
+
+resource "test_instance" "foo" {
+  ami = "bar"
+}

internal/command/workdir/statestore_config_state.go

@@ -114,7 +114,7 @@ func (s *StateStoreConfigState) SetConfig(val cty.Value, schema *configschema.Bl
 // encode the state store-specific configuration settings.
 func (s *StateStoreConfigState) PlanData(storeSchema *configschema.Block, providerSchema *configschema.Block, workspaceName string) (*plans.StateStore, error) {
 	if s == nil {
-		return nil, nil
+		panic("PlanData called on a nil *StateStoreConfigState receiver. This is a bug in Terraform and should be reported.")
 	}
 
 	if err := s.Validate(); err != nil {

internal/plans/planfile/tfplan.go

@@ -229,10 +229,7 @@ func readTfplan(r io.Reader) (*plans.Plan, error) {
 		}
 	case rawPlan.StateStore != nil:
 		rawStateStore := rawPlan.StateStore
-		config, err := valueFromTfplan(rawStateStore.Config)
-		if err != nil {
-			return nil, fmt.Errorf("plan file has invalid state_store configuration: %s", err)
-		}
+
 		provider := &plans.Provider{}
 		err = provider.SetSource(rawStateStore.Provider.Source)
 		if err != nil {
@@ -242,11 +239,21 @@ func readTfplan(r io.Reader) (*plans.Plan, error) {
 		if err != nil {
 			return nil, fmt.Errorf("plan file has invalid state_store provider version: %s", err)
 		}
+		providerConfig, err := valueFromTfplan(rawStateStore.Provider.Config)
+		if err != nil {
+			return nil, fmt.Errorf("plan file has invalid state_store configuration: %s", err)
+		}
+		provider.Config = providerConfig
+
+		storeConfig, err := valueFromTfplan(rawStateStore.Config)
+		if err != nil {
+			return nil, fmt.Errorf("plan file has invalid state_store configuration: %s", err)
+		}
 
 		plan.StateStore = &plans.StateStore{
 			Type:      rawStateStore.Type,
 			Provider:  provider,
-			Config:    config,
+			Config:    storeConfig,
 			Workspace: rawStateStore.Workspace,
 		}
 	}
@@ -759,6 +766,7 @@ func writeTfplan(plan *plans.Plan, w io.Writer) error {
 			Provider: &planproto.Provider{
 				Version: plan.StateStore.Provider.Version.String(),
 				Source:  plan.StateStore.Provider.Source.String(),
+				Config:  valueToTfplan(plan.StateStore.Provider.Config),
 			},
 			Config:    valueToTfplan(plan.StateStore.Config),
 			Workspace: plan.StateStore.Workspace,

internal/plans/planfile/tfplan_test.go

@@ -57,7 +57,13 @@ func TestTFPlanRoundTrip(t *testing.T) {
 							Namespace: "foobar",
 							Type:      "foo",
 						},
+						// Imagining a provider that has nothing in its schema
+						Config: mustNewDynamicValue(
+							cty.EmptyObjectVal,
+							cty.Object(nil),
+						),
 					},
+					// Imagining a state store with a field called `foo` in its schema
 					Config: mustNewDynamicValue(
 						cty.ObjectVal(map[string]cty.Value{
 							"foo": cty.StringVal("bar"),
@@ -136,6 +142,14 @@ func Test_writeTfplan_validation(t *testing.T) {
 							Namespace: "foobar",
 							Type:      "foo",
 						},
+						Config: mustNewDynamicValue(
+							cty.ObjectVal(map[string]cty.Value{
+								"foo": cty.StringVal("bar"),
+							}),
+							cty.Object(map[string]cty.Type{
+								"foo": cty.String,
+							}),
+						),
 					},
 					Config: mustNewDynamicValue(
 						cty.ObjectVal(map[string]cty.Value{

internal/plans/planproto/planfile.pb.go

@@ -900,6 +900,7 @@ type Provider struct {
 	state         protoimpl.MessageState `protogen:"open.v1"`
 	Source        string                 `protobuf:"bytes,1,opt,name=source,proto3" json:"source,omitempty"`
 	Version       string                 `protobuf:"bytes,2,opt,name=version,proto3" json:"version,omitempty"`
+	Config        *DynamicValue          `protobuf:"bytes,3,opt,name=config,proto3" json:"config,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
@@ -948,6 +949,13 @@ func (x *Provider) GetVersion() string {
 	return ""
 }
 
+func (x *Provider) GetConfig() *DynamicValue {
+	if x != nil {
+		return x.Config
+	}
+	return nil
+}
+
 // Change represents a change made to some object, transforming it from an old
 // state to a new state.
 type Change struct {
@@ -2265,10 +2273,11 @@ const file_planfile_proto_rawDesc = "" +
 	"\x04type\x18\x01 \x01(\tR\x04type\x12,\n" +
 	"\x06config\x18\x02 \x01(\v2\x14.tfplan.DynamicValueR\x06config\x12\x1c\n" +
 	"\tworkspace\x18\x03 \x01(\tR\tworkspace\x12,\n" +
-	"\bprovider\x18\x04 \x01(\v2\x10.tfplan.ProviderR\bprovider\"<\n" +
+	"\bprovider\x18\x04 \x01(\v2\x10.tfplan.ProviderR\bprovider\"j\n" +
 	"\bProvider\x12\x16\n" +
 	"\x06source\x18\x01 \x01(\tR\x06source\x12\x18\n" +
-	"\aversion\x18\x02 \x01(\tR\aversion\"\xbc\x03\n" +
+	"\aversion\x18\x02 \x01(\tR\aversion\x12,\n" +
+	"\x06config\x18\x03 \x01(\v2\x14.tfplan.DynamicValueR\x06config\"\xbc\x03\n" +
 	"\x06Change\x12&\n" +
 	"\x06action\x18\x01 \x01(\x0e2\x0e.tfplan.ActionR\x06action\x12,\n" +
 	"\x06values\x18\x02 \x03(\v2\x14.tfplan.DynamicValueR\x06values\x12B\n" +
@@ -2476,42 +2485,43 @@ var file_planfile_proto_depIdxs = []int32{
 	18, // 13: tfplan.Backend.config:type_name -> tfplan.DynamicValue
 	18, // 14: tfplan.StateStore.config:type_name -> tfplan.DynamicValue
 	10, // 15: tfplan.StateStore.provider:type_name -> tfplan.Provider
-	1,  // 16: tfplan.Change.action:type_name -> tfplan.Action
-	18, // 17: tfplan.Change.values:type_name -> tfplan.DynamicValue
-	19, // 18: tfplan.Change.before_sensitive_paths:type_name -> tfplan.Path
-	19, // 19: tfplan.Change.after_sensitive_paths:type_name -> tfplan.Path
-	20, // 20: tfplan.Change.importing:type_name -> tfplan.Importing
-	18, // 21: tfplan.Change.before_identity:type_name -> tfplan.DynamicValue
-	18, // 22: tfplan.Change.after_identity:type_name -> tfplan.DynamicValue
-	11, // 23: tfplan.ResourceInstanceChange.change:type_name -> tfplan.Change
-	19, // 24: tfplan.ResourceInstanceChange.required_replace:type_name -> tfplan.Path
-	2,  // 25: tfplan.ResourceInstanceChange.action_reason:type_name -> tfplan.ResourceInstanceActionReason
-	21, // 26: tfplan.DeferredResourceInstanceChange.deferred:type_name -> tfplan.Deferred
-	12, // 27: tfplan.DeferredResourceInstanceChange.change:type_name -> tfplan.ResourceInstanceChange
-	21, // 28: tfplan.DeferredActionInvocation.deferred:type_name -> tfplan.Deferred
-	22, // 29: tfplan.DeferredActionInvocation.action_invocation:type_name -> tfplan.ActionInvocationInstance
-	11, // 30: tfplan.OutputChange.change:type_name -> tfplan.Change
-	6,  // 31: tfplan.CheckResults.kind:type_name -> tfplan.CheckResults.ObjectKind
-	5,  // 32: tfplan.CheckResults.status:type_name -> tfplan.CheckResults.Status
-	28, // 33: tfplan.CheckResults.objects:type_name -> tfplan.CheckResults.ObjectResult
-	29, // 34: tfplan.Path.steps:type_name -> tfplan.Path.Step
-	18, // 35: tfplan.Importing.identity:type_name -> tfplan.DynamicValue
-	3,  // 36: tfplan.Deferred.reason:type_name -> tfplan.DeferredReason
-	18, // 37: tfplan.ActionInvocationInstance.config_value:type_name -> tfplan.DynamicValue
-	19, // 38: tfplan.ActionInvocationInstance.sensitive_config_paths:type_name -> tfplan.Path
-	23, // 39: tfplan.ActionInvocationInstance.lifecycle_action_trigger:type_name -> tfplan.LifecycleActionTrigger
-	24, // 40: tfplan.ActionInvocationInstance.invoke_action_trigger:type_name -> tfplan.InvokeActionTrigger
-	4,  // 41: tfplan.LifecycleActionTrigger.trigger_event:type_name -> tfplan.ActionTriggerEvent
-	11, // 42: tfplan.ResourceInstanceActionChange.change:type_name -> tfplan.Change
-	18, // 43: tfplan.Plan.VariablesEntry.value:type_name -> tfplan.DynamicValue
-	19, // 44: tfplan.Plan.resource_attr.attr:type_name -> tfplan.Path
-	5,  // 45: tfplan.CheckResults.ObjectResult.status:type_name -> tfplan.CheckResults.Status
-	18, // 46: tfplan.Path.Step.element_key:type_name -> tfplan.DynamicValue
-	47, // [47:47] is the sub-list for method output_type
-	47, // [47:47] is the sub-list for method input_type
-	47, // [47:47] is the sub-list for extension type_name
-	47, // [47:47] is the sub-list for extension extendee
-	0,  // [0:47] is the sub-list for field type_name
+	18, // 16: tfplan.Provider.config:type_name -> tfplan.DynamicValue
+	1,  // 17: tfplan.Change.action:type_name -> tfplan.Action
+	18, // 18: tfplan.Change.values:type_name -> tfplan.DynamicValue
+	19, // 19: tfplan.Change.before_sensitive_paths:type_name -> tfplan.Path
+	19, // 20: tfplan.Change.after_sensitive_paths:type_name -> tfplan.Path
+	20, // 21: tfplan.Change.importing:type_name -> tfplan.Importing
+	18, // 22: tfplan.Change.before_identity:type_name -> tfplan.DynamicValue
+	18, // 23: tfplan.Change.after_identity:type_name -> tfplan.DynamicValue
+	11, // 24: tfplan.ResourceInstanceChange.change:type_name -> tfplan.Change
+	19, // 25: tfplan.ResourceInstanceChange.required_replace:type_name -> tfplan.Path
+	2,  // 26: tfplan.ResourceInstanceChange.action_reason:type_name -> tfplan.ResourceInstanceActionReason
+	21, // 27: tfplan.DeferredResourceInstanceChange.deferred:type_name -> tfplan.Deferred
+	12, // 28: tfplan.DeferredResourceInstanceChange.change:type_name -> tfplan.ResourceInstanceChange
+	21, // 29: tfplan.DeferredActionInvocation.deferred:type_name -> tfplan.Deferred
+	22, // 30: tfplan.DeferredActionInvocation.action_invocation:type_name -> tfplan.ActionInvocationInstance
+	11, // 31: tfplan.OutputChange.change:type_name -> tfplan.Change
+	6,  // 32: tfplan.CheckResults.kind:type_name -> tfplan.CheckResults.ObjectKind
+	5,  // 33: tfplan.CheckResults.status:type_name -> tfplan.CheckResults.Status
+	28, // 34: tfplan.CheckResults.objects:type_name -> tfplan.CheckResults.ObjectResult
+	29, // 35: tfplan.Path.steps:type_name -> tfplan.Path.Step
+	18, // 36: tfplan.Importing.identity:type_name -> tfplan.DynamicValue
+	3,  // 37: tfplan.Deferred.reason:type_name -> tfplan.DeferredReason
+	18, // 38: tfplan.ActionInvocationInstance.config_value:type_name -> tfplan.DynamicValue
+	19, // 39: tfplan.ActionInvocationInstance.sensitive_config_paths:type_name -> tfplan.Path
+	23, // 40: tfplan.ActionInvocationInstance.lifecycle_action_trigger:type_name -> tfplan.LifecycleActionTrigger
+	24, // 41: tfplan.ActionInvocationInstance.invoke_action_trigger:type_name -> tfplan.InvokeActionTrigger
+	4,  // 42: tfplan.LifecycleActionTrigger.trigger_event:type_name -> tfplan.ActionTriggerEvent
+	11, // 43: tfplan.ResourceInstanceActionChange.change:type_name -> tfplan.Change
+	18, // 44: tfplan.Plan.VariablesEntry.value:type_name -> tfplan.DynamicValue
+	19, // 45: tfplan.Plan.resource_attr.attr:type_name -> tfplan.Path
+	5,  // 46: tfplan.CheckResults.ObjectResult.status:type_name -> tfplan.CheckResults.Status
+	18, // 47: tfplan.Path.Step.element_key:type_name -> tfplan.DynamicValue
+	48, // [48:48] is the sub-list for method output_type
+	48, // [48:48] is the sub-list for method input_type
+	48, // [48:48] is the sub-list for extension type_name
+	48, // [48:48] is the sub-list for extension extendee
+	0,  // [0:48] is the sub-list for field type_name
 }
 
 func init() { file_planfile_proto_init() }

internal/plans/planproto/planfile.proto

@@ -169,6 +169,7 @@ message StateStore {
 message Provider {
     string source = 1;
     string version = 2;
+    DynamicValue config = 3;
 }
 
 // Action describes the type of action planned for an object.