PSS: Allow pluggable state store configuration to be read from a plan file (#37957)

* feat: Allow reading state store configuration from a planfile and using it to prepare a Local backend that uses the state store

* test: Assert that we can get and use state store configuration from a plan file

* test: Add integration test showing that an apply command can use a plan file to configure and use a state store

* test: Add E2E test showing pluggable state storage being used with the full init-plan-apply workflow

* feat: A plan file will report the state storage provider among its required providers, if PSS is in use.

See the code comment added in this commit. This addition does not impact an apply command as the missing provider will be detected before this code is executed. However I'm making this change so that the method is still accurate is being able to return a complete list of providers needed by the plan.

* fix: Include error messages when there is a problem parsing provider or state store config when getting a backend from a planfile

* feat: Add trace logs to BackendForLocalPlan indicating when the provider is launched and the state store is configured

* chore: Small grammar change in error diagnostic

* refactor: Remove suggestions when the plan's state store doesn't match the implementations in the provider

* test: Add test coverage of what happens when the contents of a plan file using PSS doesn't match the resources available in the project

internal/command/apply.go

@@ -210,17 +210,17 @@ func (c *ApplyCommand) PrepareBackend(planFile *planfile.WrappedPlanFile, args *
 			))
 			return nil, diags
 		}
-		if plan.Backend == nil {
+
+		if plan.Backend == nil && plan.StateStore == nil {
 			// Should never happen; always indicates a bug in the creation of the plan file
 			diags = diags.Append(tfdiags.Sourceless(
 				tfdiags.Error,
 				"Failed to read plan from plan file",
-				"The given plan file does not have a valid backend configuration. This is a bug in the Terraform command that generated this plan file.",
+				"The given plan file has neither a valid backend nor state store configuration. This is a bug in the Terraform command that generated this plan file.",
 			))
 			return nil, diags
 		}
-		// TODO: Update BackendForLocalPlan to use state storage, and plan to be able to contain State Store config details
-		be, beDiags = c.BackendForLocalPlan(*plan.Backend)
+		be, beDiags = c.BackendForLocalPlan(plan)
 	} else {
 
 		// Load the backend

internal/command/apply_test.go

@@ -19,6 +19,8 @@ import (
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
 	"github.com/hashicorp/cli"
+	version "github.com/hashicorp/go-version"
+	tfaddr "github.com/hashicorp/terraform-registry-address"
 	"github.com/zclconf/go-cty/cty"
 
 	"github.com/hashicorp/terraform/internal/addrs"
@@ -701,6 +703,269 @@ func TestApply_plan(t *testing.T) {
 	}
 }
 
+// Test the ability to apply a plan file with a state store.
+//
+// The state store's details (provider, config, etc) are supplied by the plan,
+// which allows this test to not use any configuration.
+func TestApply_plan_stateStore(t *testing.T) {
+	// Disable test mode so input would be asked
+	test = false
+	defer func() { test = true }()
+
+	// Set some default reader/writers for the inputs
+	defaultInputReader = new(bytes.Buffer)
+	defaultInputWriter = new(bytes.Buffer)
+
+	// Create the plan file that includes a state store
+	ver := version.Must(version.NewVersion("1.2.3"))
+	providerCfg := cty.ObjectVal(map[string]cty.Value{
+		"region": cty.StringVal("spain"),
+	})
+	providerCfgRaw, err := plans.NewDynamicValue(providerCfg, providerCfg.Type())
+	if err != nil {
+		t.Fatal(err)
+	}
+	storeCfg := cty.ObjectVal(map[string]cty.Value{
+		"value": cty.StringVal("foobar"),
+	})
+	storeCfgRaw, err := plans.NewDynamicValue(storeCfg, storeCfg.Type())
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	plan := &plans.Plan{
+		Changes: plans.NewChangesSrc(),
+
+		// We'll default to the fake plan being both applyable and complete,
+		// since that's what most tests expect. Tests can override these
+		// back to false again afterwards if they need to.
+		Applyable: true,
+		Complete:  true,
+
+		StateStore: &plans.StateStore{
+			Type: "test_store",
+			Provider: &plans.Provider{
+				Version: ver,
+				Source: &tfaddr.Provider{
+					Hostname:  tfaddr.DefaultProviderRegistryHost,
+					Namespace: "hashicorp",
+					Type:      "test",
+				},
+				Config: providerCfgRaw,
+			},
+			Config:    storeCfgRaw,
+			Workspace: "default",
+		},
+	}
+
+	// Create a plan file on disk
+	//
+	// In this process we create a plan file describing the creation of a test_instance.foo resource.
+	state := testState() // State describes
+	_, snap := testModuleWithSnapshot(t, "apply")
+	planPath := testPlanFile(t, snap, state, plan)
+
+	// Create a mock, to be used as the pluggable state store described in the planfile
+	mock := testStateStoreMockWithChunkNegotiation(t, 1000)
+	view, done := testView(t)
+	c := &ApplyCommand{
+		Meta: Meta{
+			testingOverrides: &testingOverrides{
+				Providers: map[addrs.Provider]providers.Factory{
+					addrs.NewDefaultProvider("test"): providers.FactoryFixed(mock),
+				},
+			},
+			View: view,
+		},
+	}
+
+	args := []string{
+		planPath,
+	}
+	code := c.Run(args)
+	output := done(t)
+	if code != 0 {
+		t.Fatalf("bad: %d\n\n%s", code, output.Stderr())
+	}
+
+	if !mock.WriteStateBytesCalled {
+		t.Fatal("expected the test to write new state when applying the plan, but WriteStateBytesCalled is false on the mock provider.")
+	}
+}
+
+// Test unhappy paths when applying a plan file describing a state store.
+func TestApply_plan_stateStore_errorCases(t *testing.T) {
+	// Disable test mode so input would be asked
+	test = false
+	defer func() { test = true }()
+
+	t.Run("error when the provider doesn't include the state store named in the plan", func(t *testing.T) {
+		// Set some default reader/writers for the inputs
+		defaultInputReader = new(bytes.Buffer)
+		defaultInputWriter = new(bytes.Buffer)
+
+		// Create the plan file that includes a state store
+		ver := version.Must(version.NewVersion("1.2.3"))
+		providerCfg := cty.ObjectVal(map[string]cty.Value{
+			"region": cty.StringVal("spain"),
+		})
+		providerCfgRaw, err := plans.NewDynamicValue(providerCfg, providerCfg.Type())
+		if err != nil {
+			t.Fatal(err)
+		}
+		storeCfg := cty.ObjectVal(map[string]cty.Value{
+			"value": cty.StringVal("foobar"),
+		})
+		storeCfgRaw, err := plans.NewDynamicValue(storeCfg, storeCfg.Type())
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		plan := &plans.Plan{
+			Changes: plans.NewChangesSrc(),
+
+			// We'll default to the fake plan being both applyable and complete,
+			// since that's what most tests expect. Tests can override these
+			// back to false again afterwards if they need to.
+			Applyable: true,
+			Complete:  true,
+
+			StateStore: &plans.StateStore{
+				Type: "test_doesnt_exist", // Mismatched with test_store in the provider
+				Provider: &plans.Provider{
+					Version: ver,
+					Source: &tfaddr.Provider{
+						Hostname:  tfaddr.DefaultProviderRegistryHost,
+						Namespace: "hashicorp",
+						Type:      "test",
+					},
+					Config: providerCfgRaw,
+				},
+				Config:    storeCfgRaw,
+				Workspace: "default",
+			},
+		}
+
+		// Create a plan file on disk
+		//
+		// In this process we create a plan file describing the creation of a test_instance.foo resource.
+		state := testState() // State describes
+		_, snap := testModuleWithSnapshot(t, "apply")
+		planPath := testPlanFile(t, snap, state, plan)
+
+		// Create a mock, to be used as the pluggable state store described in the planfile
+		mock := testStateStoreMockWithChunkNegotiation(t, 1000)
+		view, done := testView(t)
+		c := &ApplyCommand{
+			Meta: Meta{
+				testingOverrides: &testingOverrides{
+					Providers: map[addrs.Provider]providers.Factory{
+						addrs.NewDefaultProvider("test"): providers.FactoryFixed(mock),
+					},
+				},
+				View: view,
+			},
+		}
+
+		args := []string{
+			planPath,
+			"-no-color",
+		}
+		code := c.Run(args)
+		output := done(t)
+		if code != 1 {
+			t.Fatalf("expected an error but got none: %d\n\n%s", code, output.Stdout())
+		}
+		expectedErr := "Error: State store not implemented by the provider"
+		if !strings.Contains(output.Stderr(), expectedErr) {
+			t.Fatalf("expected error to include %q, but got:\n%s", expectedErr, output.Stderr())
+		}
+	})
+
+	t.Run("error when the provider doesn't implement state stores", func(t *testing.T) {
+		// Set some default reader/writers for the inputs
+		defaultInputReader = new(bytes.Buffer)
+		defaultInputWriter = new(bytes.Buffer)
+
+		// Create the plan file that includes a state store
+		ver := version.Must(version.NewVersion("1.2.3"))
+		providerCfg := cty.ObjectVal(map[string]cty.Value{
+			"region": cty.StringVal("spain"),
+		})
+		providerCfgRaw, err := plans.NewDynamicValue(providerCfg, providerCfg.Type())
+		if err != nil {
+			t.Fatal(err)
+		}
+		storeCfg := cty.ObjectVal(map[string]cty.Value{
+			"value": cty.StringVal("foobar"),
+		})
+		storeCfgRaw, err := plans.NewDynamicValue(storeCfg, storeCfg.Type())
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		plan := &plans.Plan{
+			Changes: plans.NewChangesSrc(),
+
+			// We'll default to the fake plan being both applyable and complete,
+			// since that's what most tests expect. Tests can override these
+			// back to false again afterwards if they need to.
+			Applyable: true,
+			Complete:  true,
+
+			StateStore: &plans.StateStore{
+				Type: "test_store",
+				Provider: &plans.Provider{
+					Version: ver,
+					Source: &tfaddr.Provider{
+						Hostname:  tfaddr.DefaultProviderRegistryHost,
+						Namespace: "hashicorp",
+						Type:      "test",
+					},
+					Config: providerCfgRaw,
+				},
+				Config:    storeCfgRaw,
+				Workspace: "default",
+			},
+		}
+
+		// Create a plan file on disk
+		//
+		// In this process we create a plan file describing the creation of a test_instance.foo resource.
+		state := testState() // State describes
+		_, snap := testModuleWithSnapshot(t, "apply")
+		planPath := testPlanFile(t, snap, state, plan)
+
+		// Create a mock, to be used as the pluggable state store described in the planfile
+		mock := &testing_provider.MockProvider{}
+		view, done := testView(t)
+		c := &ApplyCommand{
+			Meta: Meta{
+				testingOverrides: &testingOverrides{
+					Providers: map[addrs.Provider]providers.Factory{
+						addrs.NewDefaultProvider("test"): providers.FactoryFixed(mock),
+					},
+				},
+				View: view,
+			},
+		}
+
+		args := []string{
+			planPath,
+			"-no-color",
+		}
+		code := c.Run(args)
+		output := done(t)
+		if code != 1 {
+			t.Fatalf("expected an error but got none: %d\n\n%s", code, output.Stdout())
+		}
+		expectedErr := "Error: Provider does not support pluggable state storage"
+		if !strings.Contains(output.Stderr(), expectedErr) {
+			t.Fatalf("expected error to include %q, but got:\n%s", expectedErr, output.Stderr())
+		}
+	})
+}
+
 func TestApply_plan_backup(t *testing.T) {
 	statePath := testTempFile(t)
 	backupPath := testTempFile(t)

internal/command/e2etest/primary_test.go

@@ -314,6 +314,88 @@ func TestPrimary_stateStore(t *testing.T) {
 	}
 }
 
+func TestPrimary_stateStore_planFile(t *testing.T) {
+
+	if !canRunGoBuild {
+		// We're running in a separate-build-then-run context, so we can't
+		// currently execute this test which depends on being able to build
+		// new executable at runtime.
+		//
+		// (See the comment on canRunGoBuild's declaration for more information.)
+		t.Skip("can't run without building a new provider executable")
+	}
+
+	t.Setenv(e2e.TestExperimentFlag, "true")
+	terraformBin := e2e.GoBuild("github.com/hashicorp/terraform", "terraform")
+
+	fixturePath := filepath.Join("testdata", "full-workflow-with-state-store-fs")
+	tf := e2e.NewBinary(t, terraformBin, fixturePath)
+
+	// In order to test integration with PSS we need a provider plugin implementing a state store.
+	// Here will build the simple6 (built with protocol v6) provider, which implements PSS.
+	simple6Provider := filepath.Join(tf.WorkDir(), "terraform-provider-simple6")
+	simple6ProviderExe := e2e.GoBuild("github.com/hashicorp/terraform/internal/provider-simple-v6/main", simple6Provider)
+
+	// Move the provider binaries into a directory that we will point terraform
+	// to using the -plugin-dir cli flag.
+	platform := getproviders.CurrentPlatform.String()
+	hashiDir := "cache/registry.terraform.io/hashicorp/"
+	if err := os.MkdirAll(tf.Path(hashiDir, "simple6/0.0.1/", platform), os.ModePerm); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.Rename(simple6ProviderExe, tf.Path(hashiDir, "simple6/0.0.1/", platform, "terraform-provider-simple6")); err != nil {
+		t.Fatal(err)
+	}
+
+	//// INIT
+	stdout, stderr, err := tf.Run("init", "-enable-pluggable-state-storage-experiment=true", "-plugin-dir=cache", "-no-color")
+	if err != nil {
+		t.Fatalf("unexpected init error: %s\nstderr:\n%s", err, stderr)
+	}
+
+	if !strings.Contains(stdout, "Terraform created an empty state file for the default workspace") {
+		t.Errorf("notice about creating the default workspace is missing from init output:\n%s", stdout)
+	}
+
+	//// PLAN
+	planFile := "testplan"
+	_, stderr, err = tf.Run("plan", "-out="+planFile, "-no-color")
+	if err != nil {
+		t.Fatalf("unexpected apply error: %s\nstderr:\n%s", err, stderr)
+	}
+
+	//// APPLY
+	stdout, stderr, err = tf.Run("apply", "-auto-approve", "-no-color", planFile)
+	if err != nil {
+		t.Fatalf("unexpected apply error: %s\nstderr:\n%s", err, stderr)
+	}
+
+	if !strings.Contains(stdout, "Resources: 1 added, 0 changed, 0 destroyed") {
+		t.Errorf("incorrect apply tally; want 1 added:\n%s", stdout)
+	}
+
+	// Check the statefile saved by the fs state store.
+	path := "states/default/terraform.tfstate"
+	f, err := tf.OpenFile(path)
+	if err != nil {
+		t.Fatalf("unexpected error opening state file %s: %s\nstderr:\n%s", path, err, stderr)
+	}
+	defer f.Close()
+
+	stateFile, err := statefile.Read(f)
+	if err != nil {
+		t.Fatalf("unexpected error reading statefile %s: %s\nstderr:\n%s", path, err, stderr)
+	}
+
+	r := stateFile.State.RootModule().Resources
+	if len(r) != 1 {
+		t.Fatalf("expected state to include one resource, but got %d", len(r))
+	}
+	if _, ok := r["terraform_data.my-data"]; !ok {
+		t.Fatalf("expected state to include terraform_data.my-data but it's missing")
+	}
+}
+
 func TestPrimary_stateStore_inMem(t *testing.T) {
 	if !canRunGoBuild {
 		// We're running in a separate-build-then-run context, so we can't

internal/command/meta_backend.go

@@ -327,40 +327,217 @@ func (m *Meta) selectWorkspace(b backend.Backend) error {
 	return m.SetWorkspace(workspace)
 }
 
-// BackendForLocalPlan is similar to Backend, but uses backend settings that were
-// stored in a plan.
+// BackendForLocalPlan is similar to Backend, but uses settings that were
+// stored in a plan when preparing the returned operations backend.
+// The plan's data may describe `backend` or `state_store` configuration.
 //
 // The current workspace name is also stored as part of the plan, and so this
 // method will check that it matches the currently-selected workspace name
 // and produce error diagnostics if not.
-func (m *Meta) BackendForLocalPlan(settings plans.Backend) (backendrun.OperationsBackend, tfdiags.Diagnostics) {
+func (m *Meta) BackendForLocalPlan(plan *plans.Plan) (backendrun.OperationsBackend, tfdiags.Diagnostics) {
 	var diags tfdiags.Diagnostics
 
-	f := backendInit.Backend(settings.Type)
-	if f == nil {
-		diags = diags.Append(errBackendSavedUnknown{settings.Type})
-		return nil, diags
-	}
-	b := f()
-	log.Printf("[TRACE] Meta.BackendForLocalPlan: instantiated backend of type %T", b)
+	var b backend.Backend
+	switch {
+	case plan.StateStore != nil:
+		settings := plan.StateStore
+
+		// BackendForLocalPlan is used in the context of an apply command using a plan file,
+		// so we can read locks directly from the lock file and trust it contains what we need.
+		locks, lockDiags := m.lockedDependencies()
+		diags = diags.Append(lockDiags)
+		if lockDiags.HasErrors() {
+			return nil, diags
+		}
 
-	schema := b.ConfigSchema()
-	configVal, err := settings.Config.Decode(schema.ImpliedType())
-	if err != nil {
-		diags = diags.Append(fmt.Errorf("saved backend configuration is invalid: %w", err))
-		return nil, diags
-	}
+		factories, err := m.ProviderFactoriesFromLocks(locks)
+		if err != nil {
+			// This may happen if the provider isn't present in the provider cache.
+			// This should be caught earlier by logic that diffs the config against the backend state file.
+			return nil, diags.Append(&hcl.Diagnostic{
+				Severity: hcl.DiagError,
+				Summary:  "Provider unavailable",
+				Detail: fmt.Sprintf("Terraform experienced an error when trying to use provider %s (%q) to initialize the %q state store: %s",
+					settings.Provider.Source.Type,
+					settings.Provider.Source,
+					settings.Type,
+					err),
+			})
+		}
 
-	newVal, validateDiags := b.PrepareConfig(configVal)
-	diags = diags.Append(validateDiags)
-	if validateDiags.HasErrors() {
-		return nil, diags
-	}
+		factory, exists := factories[*settings.Provider.Source]
+		if !exists {
+			return nil, diags.Append(&hcl.Diagnostic{
+				Severity: hcl.DiagError,
+				Summary:  "Provider unavailable",
+				Detail: fmt.Sprintf("The provider %s (%q) is required to initialize the %q state store, but the matching provider factory is missing. This is a bug in Terraform and should be reported.",
+					settings.Provider.Source.Type,
+					settings.Provider.Source,
+					settings.Type,
+				),
+			})
+		}
 
-	configureDiags := b.Configure(newVal)
-	diags = diags.Append(configureDiags)
-	if configureDiags.HasErrors() {
-		return nil, diags
+		provider, err := factory()
+		if err != nil {
+			diags = diags.Append(fmt.Errorf("error when obtaining provider instance during state store initialization: %w", err))
+			return nil, diags
+		}
+		log.Printf("[TRACE] Meta.BackendForLocalPlan: launched instance of provider %s (%q)",
+			settings.Provider.Source.Type,
+			settings.Provider.Source,
+		)
+
+		// We purposefully don't have a deferred call to the provider's Close method here because the calling code needs a
+		// running provider instance inside the returned backend.Backend instance.
+		// Stopping the provider process is the responsibility of the calling code.
+
+		resp := provider.GetProviderSchema()
+
+		if len(resp.StateStores) == 0 {
+			diags = diags.Append(&hcl.Diagnostic{
+				Severity: hcl.DiagError,
+				Summary:  "Provider does not support pluggable state storage",
+				Detail: fmt.Sprintf("There are no state stores implemented by provider %s (%q)",
+					settings.Provider.Source.Type,
+					settings.Provider.Source),
+			})
+			return nil, diags
+		}
+
+		stateStoreSchema, exists := resp.StateStores[settings.Type]
+		if !exists {
+			diags = diags.Append(&hcl.Diagnostic{
+				Severity: hcl.DiagError,
+				Summary:  "State store not implemented by the provider",
+				Detail: fmt.Sprintf("State store %q is not implemented by provider %s (%q)",
+					settings.Type,
+					settings.Provider.Source.Type,
+					settings.Provider.Source,
+				),
+			})
+			return nil, diags
+		}
+
+		// Get the provider config from the backend state file.
+		providerConfigVal, err := settings.Provider.Config.Decode(resp.Provider.Body.ImpliedType())
+		if err != nil {
+			diags = diags.Append(
+				&hcl.Diagnostic{
+					Severity: hcl.DiagError,
+					Summary:  "Error reading provider configuration state",
+					Detail: fmt.Sprintf("Terraform experienced an error reading provider configuration for provider %s (%q) while configuring state store %s: %s",
+						settings.Provider.Source.Type,
+						settings.Provider.Source,
+						settings.Type,
+						err,
+					),
+				},
+			)
+			return nil, diags
+		}
+
+		// Get the state store config from the backend state file.
+		stateStoreConfigVal, err := settings.Config.Decode(stateStoreSchema.Body.ImpliedType())
+		if err != nil {
+			diags = diags.Append(
+				&hcl.Diagnostic{
+					Severity: hcl.DiagError,
+					Summary:  "Error reading state store configuration state",
+					Detail: fmt.Sprintf("Terraform experienced an error reading state store configuration for state store %s in provider %s (%q): %s",
+						settings.Type,
+						settings.Provider.Source.Type,
+						settings.Provider.Source,
+						err,
+					),
+				},
+			)
+			return nil, diags
+		}
+
+		// Validate and configure the provider
+		//
+		// NOTE: there are no marks we need to remove at this point.
+		// We haven't added marks since the provider config from the backend state was used
+		// because the state-storage provider's config isn't going to be presented to the user via terminal output or diags.
+		validateResp := provider.ValidateProviderConfig(providers.ValidateProviderConfigRequest{
+			Config: providerConfigVal,
+		})
+		diags = diags.Append(validateResp.Diagnostics)
+		if diags.HasErrors() {
+			return nil, diags
+		}
+
+		configureResp := provider.ConfigureProvider(providers.ConfigureProviderRequest{
+			TerraformVersion: tfversion.SemVer.String(),
+			Config:           providerConfigVal,
+		})
+		diags = diags.Append(configureResp.Diagnostics)
+		if diags.HasErrors() {
+			return nil, diags
+		}
+
+		// Now that the provider is configured we can begin using the state store through
+		// the backend.Backend interface.
+		p, err := backendPluggable.NewPluggable(provider, settings.Type)
+		if err != nil {
+			diags = diags.Append(err)
+			return nil, diags
+		}
+
+		// Validate and configure the state store
+		//
+		// Note: we do not use the value returned from PrepareConfig for state stores,
+		// however that old approach is still used with backends for compatibility reasons.
+		_, validateDiags := p.PrepareConfig(stateStoreConfigVal)
+		diags = diags.Append(validateDiags)
+		if validateDiags.HasErrors() {
+			return nil, diags
+		}
+
+		configureDiags := p.Configure(stateStoreConfigVal)
+		diags = diags.Append(configureDiags)
+		if configureDiags.HasErrors() {
+			return nil, diags
+		}
+		log.Printf("[TRACE] Meta.BackendForLocalPlan: finished configuring state store %s in provider %s (%q)",
+			settings.Type,
+			settings.Provider.Source.Type,
+			settings.Provider.Source,
+		)
+
+		// The fully configured Pluggable is used as the instance of backend.Backend
+		b = p
+
+	default:
+		settings := plan.Backend
+
+		f := backendInit.Backend(settings.Type)
+		if f == nil {
+			diags = diags.Append(errBackendSavedUnknown{settings.Type})
+			return nil, diags
+		}
+		b = f()
+		log.Printf("[TRACE] Meta.BackendForLocalPlan: instantiated backend of type %T", b)
+
+		schema := b.ConfigSchema()
+		configVal, err := settings.Config.Decode(schema.ImpliedType())
+		if err != nil {
+			diags = diags.Append(fmt.Errorf("saved backend configuration is invalid: %w", err))
+			return nil, diags
+		}
+
+		newVal, validateDiags := b.PrepareConfig(configVal)
+		diags = diags.Append(validateDiags)
+		if validateDiags.HasErrors() {
+			return nil, diags
+		}
+
+		configureDiags := b.Configure(newVal)
+		diags = diags.Append(configureDiags)
+		if configureDiags.HasErrors() {
+			return nil, diags
+		}
 	}
 
 	// If the backend supports CLI initialization, do it.

internal/command/meta_backend_test.go

@@ -1560,7 +1560,7 @@ func TestMetaBackend_configuredBackendUnsetCopy(t *testing.T) {
 	}
 }
 
-// A plan that has uses the local backend
+// A plan that has uses the local backend and local state storage
 func TestMetaBackend_planLocal(t *testing.T) {
 	// Create a temporary working directory that is empty
 	td := t.TempDir()
@@ -1575,17 +1575,19 @@ func TestMetaBackend_planLocal(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	backendConfig := plans.Backend{
-		Type:      "local",
-		Config:    backendConfigRaw,
-		Workspace: "default",
+	plan := &plans.Plan{
+		Backend: &plans.Backend{
+			Type:      "local",
+			Config:    backendConfigRaw,
+			Workspace: "default",
+		},
 	}
 
 	// Setup the meta
 	m := testMetaBackend(t, nil)
 
 	// Get the backend
-	b, diags := m.BackendForLocalPlan(backendConfig)
+	b, diags := m.BackendForLocalPlan(plan)
 	if diags.HasErrors() {
 		t.Fatal(diags.Err())
 	}
@@ -1649,6 +1651,71 @@ func TestMetaBackend_planLocal(t *testing.T) {
 	}
 }
 
+// A plan that has uses the local backend and pluggable state storage
+func TestMetaBackend_planLocal_stateStore(t *testing.T) {
+	// Create a temporary working directory
+	td := t.TempDir()
+	testCopyDir(t, testFixturePath("state-store-unchanged"), td)
+	t.Chdir(td)
+
+	stateStoreConfigBlock := cty.ObjectVal(map[string]cty.Value{
+		"value": cty.StringVal("foobar"),
+	})
+	stateStoreConfigRaw, err := plans.NewDynamicValue(stateStoreConfigBlock, stateStoreConfigBlock.Type())
+	if err != nil {
+		t.Fatal(err)
+	}
+	providerAddr := addrs.MustParseProviderSourceString("registry.terraform.io/hashicorp/test")
+
+	plan := &plans.Plan{
+		StateStore: &plans.StateStore{
+			Type:      "test_store",
+			Config:    stateStoreConfigRaw,
+			Workspace: backend.DefaultStateName,
+			Provider: &plans.Provider{
+				Version: version.Must(version.NewVersion("1.2.3")), // Matches lock file in the test fixtures
+				Source:  &providerAddr,
+				Config:  nil,
+			},
+		},
+	}
+
+	// Setup the meta, including a mock provider set up to mock PSS
+	m := testMetaBackend(t, nil)
+	mock := testStateStoreMockWithChunkNegotiation(t, 1000)
+	m.testingOverrides = &testingOverrides{
+		Providers: map[addrs.Provider]providers.Factory{
+			addrs.NewDefaultProvider("test"): providers.FactoryFixed(mock),
+		},
+	}
+
+	// Get the backend
+	b, diags := m.BackendForLocalPlan(plan)
+	if diags.HasErrors() {
+		t.Fatal(diags.Err())
+	}
+
+	// Check the state
+	s, sDiags := b.StateMgr(backend.DefaultStateName)
+	if sDiags.HasErrors() {
+		t.Fatalf("unexpected error: %s", sDiags.Err())
+	}
+	if err := s.RefreshState(); err != nil {
+		t.Fatalf("unexpected error: %s", err)
+	}
+	state := s.State()
+	if state != nil {
+		t.Fatalf("state should be nil: %#v", state)
+	}
+
+	// Write some state
+	state = states.NewState()
+	s.WriteState(state)
+	if err := s.PersistState(nil); err != nil {
+		t.Fatalf("unexpected error: %s", err)
+	}
+}
+
 // A plan with a custom state save path
 func TestMetaBackend_planLocalStatePath(t *testing.T) {
 	td := t.TempDir()
@@ -1666,10 +1733,12 @@ func TestMetaBackend_planLocalStatePath(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	plannedBackend := plans.Backend{
-		Type:      "local",
-		Config:    backendConfigRaw,
-		Workspace: "default",
+	plan := &plans.Plan{
+		Backend: &plans.Backend{
+			Type:      "local",
+			Config:    backendConfigRaw,
+			Workspace: "default",
+		},
 	}
 
 	// Create an alternate output path
@@ -1686,7 +1755,7 @@ func TestMetaBackend_planLocalStatePath(t *testing.T) {
 	m.stateOutPath = statePath
 
 	// Get the backend
-	b, diags := m.BackendForLocalPlan(plannedBackend)
+	b, diags := m.BackendForLocalPlan(plan)
 	if diags.HasErrors() {
 		t.Fatal(diags.Err())
 	}
@@ -1765,17 +1834,19 @@ func TestMetaBackend_planLocalMatch(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	backendConfig := plans.Backend{
-		Type:      "local",
-		Config:    backendConfigRaw,
-		Workspace: "default",
+	plan := &plans.Plan{
+		Backend: &plans.Backend{
+			Type:      "local",
+			Config:    backendConfigRaw,
+			Workspace: "default",
+		},
 	}
 
 	// Setup the meta
 	m := testMetaBackend(t, nil)
 
 	// Get the backend
-	b, diags := m.BackendForLocalPlan(backendConfig)
+	b, diags := m.BackendForLocalPlan(plan)
 	if diags.HasErrors() {
 		t.Fatal(diags.Err())
 	}

internal/plans/plan.go

@@ -181,9 +181,30 @@ func (p *Plan) ProviderAddrs() []addrs.AbsProviderConfig {
 	}
 
 	m := map[string]addrs.AbsProviderConfig{}
+
+	// Get all provider requirements from resources.
 	for _, rc := range p.Changes.Resources {
 		m[rc.ProviderAddr.String()] = rc.ProviderAddr
 	}
+
+	// Get the provider required for pluggable state storage, if that's in use.
+	//
+	// This check should be redundant as:
+	// 1) Any provider used for state storage would be in required_providers, which is checked separately elsewhere.
+	// 2) An apply operation that uses the planfile only checks the providers needed for the plan _after_ the operations backend
+	//    for the operation is set up, and that process will detect if the provider needed for state storage is missing.
+	//
+	// However, for completeness when describing the providers needed by a plan, it is included here.
+	if p.StateStore != nil {
+		address := addrs.AbsProviderConfig{
+			Module:   addrs.RootModule, // A state_store block is only ever in the root module
+			Provider: *p.StateStore.Provider.Source,
+			// Alias: aliases are not permitted when using a provider for PSS.
+		}
+
+		m[p.StateStore.Provider.Source.String()] = address
+	}
+
 	if len(m) == 0 {
 		return nil
 	}

internal/plans/plan_test.go

@@ -9,12 +9,38 @@ import (
 	"github.com/go-test/deep"
 
 	"github.com/hashicorp/terraform/internal/addrs"
+	"github.com/zclconf/go-cty/cty"
 )
 
 func TestProviderAddrs(t *testing.T) {
+	// Inputs for plan
+	provider := &Provider{}
+	err := provider.SetSource("registry.terraform.io/hashicorp/pluggable")
+	if err != nil {
+		panic(err)
+	}
+	err = provider.SetVersion("9.9.9")
+	if err != nil {
+		panic(err)
+	}
+	config, err := NewDynamicValue(cty.ObjectVal(map[string]cty.Value{
+		"foo": cty.StringVal("bar"),
+	}), cty.Object(map[string]cty.Type{
+		"foo": cty.String,
+	}))
+	if err != nil {
+		panic(err)
+	}
+	provider.Config = config
 
 	// Prepare plan
 	plan := &Plan{
+		StateStore: &StateStore{
+			Type:      "pluggable_foobar",
+			Provider:  provider,
+			Config:    config,
+			Workspace: "default",
+		},
 		VariableValues: map[string]DynamicValue{},
 		Changes: &ChangesSrc{
 			Resources: []*ResourceInstanceChangeSrc{
@@ -67,6 +93,11 @@ func TestProviderAddrs(t *testing.T) {
 			Module:   addrs.RootModule,
 			Provider: addrs.NewDefaultProvider("test"),
 		},
+		// Provider used for pluggable state storage
+		{
+			Module:   addrs.RootModule,
+			Provider: addrs.NewDefaultProvider("pluggable"),
+		},
 	}
 
 	for _, problem := range deep.Equal(got, want) {