add test and removed backend state from cloud

internal/cloud/backend.go

@@ -526,15 +526,10 @@ func (b *Cloud) DeleteWorkspace(name string) error {
 	}
 
 	// Configure the remote workspace name.
-	client := &remoteClient{
+	State := &State{tfeClient: b.client, organization: b.organization, workspace: &tfe.Workspace{
-		client:       b.client,
+		Name: name,
-		organization: b.organization,
+	}}
-		workspace: &tfe.Workspace{
+	return State.Delete()
-			Name: name,
-		},
-	}
-
-	return client.Delete()
 }
 
 // StateMgr implements backend.Enhanced.
@@ -619,16 +614,7 @@ func (b *Cloud) StateMgr(name string) (statemgr.Full, error) {
 		}
 	}
 
-	client := &remoteClient{
+	return &State{tfeClient: b.client, organization: b.organization, workspace: workspace}, nil
-		client:       b.client,
-		organization: b.organization,
-		workspace:    workspace,
-
-		// This is optionally set during Terraform Enterprise runs.
-		runID: os.Getenv("TFE_RUN_ID"),
-	}
-
-	return NewState(client), nil
 }
 
 // Operation implements backend.Enhanced.

internal/cloud/backend_state.go

@@ -1,196 +0,0 @@
-package cloud
-
-import (
-	"bytes"
-	"context"
-	"crypto/md5"
-	"encoding/base64"
-	"encoding/json"
-	"errors"
-	"fmt"
-
-	tfe "github.com/hashicorp/go-tfe"
-
-	"github.com/hashicorp/terraform/internal/command/jsonstate"
-	"github.com/hashicorp/terraform/internal/states/remote"
-	"github.com/hashicorp/terraform/internal/states/statefile"
-	"github.com/hashicorp/terraform/internal/states/statemgr"
-)
-
-type remoteClient struct {
-	client         *tfe.Client
-	lockInfo       *statemgr.LockInfo
-	organization   string
-	runID          string
-	stateUploadErr bool
-	workspace      *tfe.Workspace
-	forcePush      bool
-}
-
-// Get the remote state.
-func (r *remoteClient) Get() (*remote.Payload, error) {
-	ctx := context.Background()
-
-	sv, err := r.client.StateVersions.ReadCurrent(ctx, r.workspace.ID)
-	if err != nil {
-		if err == tfe.ErrResourceNotFound {
-			// If no state exists, then return nil.
-			return nil, nil
-		}
-		return nil, fmt.Errorf("failed to retrieve state: %w", err)
-	}
-
-	state, err := r.client.StateVersions.Download(ctx, sv.DownloadURL)
-	if err != nil {
-		return nil, fmt.Errorf("failed to download state: %w", err)
-	}
-
-	// If the state is empty, then return nil.
-	if len(state) == 0 {
-		return nil, nil
-	}
-
-	// Get the MD5 checksum of the state.
-	sum := md5.Sum(state)
-
-	return &remote.Payload{
-		Data: state,
-		MD5:  sum[:],
-	}, nil
-}
-
-// Put the remote state.
-func (r *remoteClient) Put(state []byte) error {
-	ctx := context.Background()
-
-	// Read the raw state into a Terraform state.
-	stateFile, err := statefile.Read(bytes.NewReader(state))
-	if err != nil {
-		return fmt.Errorf("failed to read state: %w", err)
-	}
-
-	ov, err := jsonstate.MarshalOutputs(stateFile.State.RootModule().OutputValues)
-	if err != nil {
-		return fmt.Errorf("failed to translate outputs: %w", err)
-	}
-	o, err := json.Marshal(ov)
-	if err != nil {
-		return fmt.Errorf("failed to marshal outputs to json: %w", err)
-	}
-
-	options := tfe.StateVersionCreateOptions{
-		Lineage:          tfe.String(stateFile.Lineage),
-		Serial:           tfe.Int64(int64(stateFile.Serial)),
-		MD5:              tfe.String(fmt.Sprintf("%x", md5.Sum(state))),
-		State:            tfe.String(base64.StdEncoding.EncodeToString(state)),
-		Force:            tfe.Bool(r.forcePush),
-		JSONStateOutputs: tfe.String(base64.StdEncoding.EncodeToString(o)),
-	}
-
-	// If we have a run ID, make sure to add it to the options
-	// so the state will be properly associated with the run.
-	if r.runID != "" {
-		options.Run = &tfe.Run{ID: r.runID}
-	}
-
-	// Create the new state.
-	_, err = r.client.StateVersions.Create(ctx, r.workspace.ID, options)
-	if err != nil {
-		r.stateUploadErr = true
-		return fmt.Errorf("failed to upload state: %w", err)
-	}
-
-	return nil
-}
-
-// Delete the remote state.
-func (r *remoteClient) Delete() error {
-	err := r.client.Workspaces.Delete(context.Background(), r.organization, r.workspace.Name)
-	if err != nil && err != tfe.ErrResourceNotFound {
-		return fmt.Errorf("failed to delete workspace %s: %w", r.workspace.Name, err)
-	}
-
-	return nil
-}
-
-// EnableForcePush to allow the remote client to overwrite state
-// by implementing remote.ClientForcePusher
-func (r *remoteClient) EnableForcePush() {
-	r.forcePush = true
-}
-
-// Lock the remote state.
-func (r *remoteClient) Lock(info *statemgr.LockInfo) (string, error) {
-	ctx := context.Background()
-
-	lockErr := &statemgr.LockError{Info: r.lockInfo}
-
-	// Lock the workspace.
-	_, err := r.client.Workspaces.Lock(ctx, r.workspace.ID, tfe.WorkspaceLockOptions{
-		Reason: tfe.String("Locked by Terraform"),
-	})
-	if err != nil {
-		if err == tfe.ErrWorkspaceLocked {
-			lockErr.Info = info
-			err = fmt.Errorf("%s (lock ID: \"%s/%s\")", err, r.organization, r.workspace.Name)
-		}
-		lockErr.Err = err
-		return "", lockErr
-	}
-
-	r.lockInfo = info
-
-	return r.lockInfo.ID, nil
-}
-
-// Unlock the remote state.
-func (r *remoteClient) Unlock(id string) error {
-	ctx := context.Background()
-
-	// We first check if there was an error while uploading the latest
-	// state. If so, we will not unlock the workspace to prevent any
-	// changes from being applied until the correct state is uploaded.
-	if r.stateUploadErr {
-		return nil
-	}
-
-	lockErr := &statemgr.LockError{Info: r.lockInfo}
-
-	// With lock info this should be treated as a normal unlock.
-	if r.lockInfo != nil {
-		// Verify the expected lock ID.
-		if r.lockInfo.ID != id {
-			lockErr.Err = errors.New("lock ID does not match existing lock")
-			return lockErr
-		}
-
-		// Unlock the workspace.
-		_, err := r.client.Workspaces.Unlock(ctx, r.workspace.ID)
-		if err != nil {
-			lockErr.Err = err
-			return lockErr
-		}
-
-		return nil
-	}
-
-	// Verify the optional force-unlock lock ID.
-	if r.organization+"/"+r.workspace.Name != id {
-		lockErr.Err = fmt.Errorf(
-			"lock ID %q does not match existing lock ID \"%s/%s\"",
-			id,
-			r.organization,
-			r.workspace.Name,
-		)
-		return lockErr
-	}
-
-	// Force unlock the workspace.
-	_, err := r.client.Workspaces.ForceUnlock(ctx, r.workspace.ID)
-	if err != nil {
-		lockErr.Err = err
-		return lockErr
-	}
-
-	return nil
-}

internal/cloud/backend_state_test.go

@@ -1,103 +0,0 @@
-package cloud
-
-import (
-	"bytes"
-	"os"
-	"testing"
-
-	tfe "github.com/hashicorp/go-tfe"
-
-	"github.com/hashicorp/terraform/internal/states"
-	"github.com/hashicorp/terraform/internal/states/remote"
-	"github.com/hashicorp/terraform/internal/states/statefile"
-)
-
-func TestRemoteClient_impl(t *testing.T) {
-	var _ remote.Client = new(remoteClient)
-}
-
-func TestRemoteClient(t *testing.T) {
-	client := testRemoteClient(t)
-	remote.TestClient(t, client)
-}
-
-func TestRemoteClient_stateVersionCreated(t *testing.T) {
-	b, bCleanup := testBackendWithName(t)
-	defer bCleanup()
-
-	raw, err := b.StateMgr(testBackendSingleWorkspaceName)
-	if err != nil {
-		t.Fatalf("error: %v", err)
-	}
-
-	client := raw.(*State).Client
-
-	err = client.Put(([]byte)(`
-{
-	"version": 4,
-	"terraform_version": "1.3.0",
-	"serial": 1,
-	"lineage": "backend-change",
-	"outputs": {
-			"foo": {
-					"type": "string",
-					"value": "bar"
-			}
-	}
-}`))
-	if err != nil {
-		t.Fatalf("expected no error, got %v", err)
-	}
-
-	stateVersionsAPI := b.client.StateVersions.(*MockStateVersions)
-	if got, want := len(stateVersionsAPI.stateVersions), 1; got != want {
-		t.Fatalf("wrong number of state versions in the mock client %d; want %d", got, want)
-	}
-
-	var stateVersion *tfe.StateVersion
-	for _, sv := range stateVersionsAPI.stateVersions {
-		stateVersion = sv
-	}
-
-	if stateVersionsAPI.outputStates[stateVersion.ID] == nil || len(stateVersionsAPI.outputStates[stateVersion.ID]) == 0 {
-		t.Fatal("no state version outputs in the mock client")
-	}
-}
-
-func TestRemoteClient_TestRemoteLocks(t *testing.T) {
-	b, bCleanup := testBackendWithName(t)
-	defer bCleanup()
-
-	s1, err := b.StateMgr(testBackendSingleWorkspaceName)
-	if err != nil {
-		t.Fatalf("expected no error, got %v", err)
-	}
-
-	s2, err := b.StateMgr(testBackendSingleWorkspaceName)
-	if err != nil {
-		t.Fatalf("expected no error, got %v", err)
-	}
-
-	remote.TestRemoteLocks(t, s1.(*State).Client, s2.(*State).Client)
-}
-
-func TestRemoteClient_withRunID(t *testing.T) {
-	// Set the TFE_RUN_ID environment variable before creating the client!
-	if err := os.Setenv("TFE_RUN_ID", GenerateID("run-")); err != nil {
-		t.Fatalf("error setting env var TFE_RUN_ID: %v", err)
-	}
-
-	// Create a new test client.
-	client := testRemoteClient(t)
-
-	// Create a new empty state.
-	sf := statefile.New(states.NewState(), "", 0)
-	var buf bytes.Buffer
-	statefile.Write(sf, &buf)
-
-	// Store the new state to verify (this will be done
-	// by the mock that is used) that the run ID is set.
-	if err := client.Put(buf.Bytes()); err != nil {
-		t.Fatalf("expected no error, got %v", err)
-	}
-}

internal/cloud/state.go

@@ -1,30 +1,59 @@
 package cloud
 
 import (
+	"bytes"
 	"context"
+	"crypto/md5"
+	"encoding/base64"
 	"encoding/json"
 	"errors"
 	"fmt"
 	"log"
+	"os"
 	"strings"
+	"sync"
 
-	"github.com/hashicorp/go-tfe"
+	tfe "github.com/hashicorp/go-tfe"
 	"github.com/zclconf/go-cty/cty"
 	"github.com/zclconf/go-cty/cty/gocty"
 
+	uuid "github.com/hashicorp/go-uuid"
+	"github.com/hashicorp/terraform/internal/command/jsonstate"
 	"github.com/hashicorp/terraform/internal/states"
 	"github.com/hashicorp/terraform/internal/states/remote"
+	"github.com/hashicorp/terraform/internal/states/statefile"
 	"github.com/hashicorp/terraform/internal/states/statemgr"
 	"github.com/hashicorp/terraform/internal/terraform"
 )
 
-// State is similar to remote State and delegates to it, except in the case of output values,
+// State implements the State interfaces in the state package to handle
-// which use a separate methodology that ensures the caller is authorized to read cloud
+// reading and writing the remote state to TFC. This State on its own does no
-// workspace outputs.
+// local caching so every persist will go to the remote storage and local
+// writes will go to memory.
 type State struct {
-	Client *remoteClient
+	mu sync.Mutex
 
-	delegate remote.State
+	// Client Client
+
+	// We track two pieces of meta data in addition to the state itself:
+	//
+	// lineage - the state's unique ID
+	// serial  - the monotonic counter of "versions" of the state
+	//
+	// Both of these (along with state) have a sister field
+	// that represents the values read in from an existing source.
+	// All three of these values are used to determine if the new
+	// state has changed from an existing state we read in.
+	lineage, readLineage string
+	serial, readSerial   uint64
+	state, readState     *states.State
+	disableLocks         bool
+	tfeClient            *tfe.Client
+	organization         string
+	workspace            *tfe.Workspace
+	stateUploadErr       bool
+	forcePush            bool
+	lockInfo             *statemgr.LockInfo
 }
 
 var ErrStateVersionUnauthorizedUpgradeState = errors.New(strings.TrimSpace(`
@@ -34,70 +63,370 @@ of authorization and therefore this error can usually be fixed by upgrading the
 remote state version.
 `))
 
-// Proof that cloud State is a statemgr.Persistent interface
+var _ statemgr.Full = (*State)(nil)
-var _ statemgr.Persistent = (*State)(nil)
+var _ statemgr.Migrator = (*State)(nil)
 
-func NewState(client *remoteClient) *State {
+// statemgr.Reader impl.
-	return &State{
-		Client:   client,
-		delegate: remote.State{Client: client},
-	}
-}
-
-// State delegates calls to read State to the remote State
 func (s *State) State() *states.State {
-	return s.delegate.State()
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	return s.state.DeepCopy()
 }
 
-// Lock delegates calls to lock state to the remote State
+// StateForMigration is part of our implementation of statemgr.Migrator.
-func (s *State) Lock(info *statemgr.LockInfo) (string, error) {
+func (s *State) StateForMigration() *statefile.File {
-	return s.delegate.Lock(info)
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	return statefile.New(s.state.DeepCopy(), s.lineage, s.serial)
 }
 
-// Unlock delegates calls to unlock state to the remote State
+// WriteStateForMigration is part of our implementation of statemgr.Migrator.
-func (s *State) Unlock(id string) error {
+func (s *State) WriteStateForMigration(f *statefile.File, force bool) error {
-	return s.delegate.Unlock(id)
+	s.mu.Lock()
-}
+	defer s.mu.Unlock()
 
-// RefreshState delegates calls to refresh State to the remote State
+	if !force {
-func (s *State) RefreshState() error {
+		checkFile := statefile.New(s.state, s.lineage, s.serial)
-	return s.delegate.RefreshState()
+		if err := statemgr.CheckValidImport(f, checkFile); err != nil {
-}
+			return err
+		}
+	}
+
+	// The remote backend needs to pass the `force` flag through to its client.
+	// For backends that support such operations, inform the client
+	// that a force push has been requested
+	if force {
+		s.EnableForcePush()
+	}
+
+	// We create a deep copy of the state here, because the caller also has
+	// a reference to the given object and can potentially go on to mutate
+	// it after we return, but we want the snapshot at this point in time.
+	s.state = f.State.DeepCopy()
+	s.lineage = f.Lineage
+	s.serial = f.Serial
 
-// RefreshState delegates calls to refresh State to the remote State
-func (s *State) PersistState(schemas *terraform.Schemas) error {
-	s.delegate.PersistState(schemas)
 	return nil
 }
 
-// WriteState delegates calls to write State to the remote State
+// DisableLocks turns the Lock and Unlock methods into no-ops. This is intended
-func (s *State) WriteState(state *states.State) error {
+// to be called during initialization of a state manager and should not be
-	return s.delegate.WriteState(state)
+// called after any of the statemgr.Full interface methods have been called.
+func (s *State) DisableLocks() {
+	s.disableLocks = true
 }
 
-func (s *State) fallbackReadOutputsFromFullState() (map[string]*states.OutputValue, error) {
+// StateSnapshotMeta returns the metadata from the most recently persisted
-	log.Printf("[DEBUG] falling back to reading full state")
+// or refreshed persistent state snapshot.
+//
+// This is an implementation of statemgr.PersistentMeta.
+func (s *State) StateSnapshotMeta() statemgr.SnapshotMeta {
+	return statemgr.SnapshotMeta{
+		Lineage: s.lineage,
+		Serial:  s.serial,
+	}
+}
 
-	if err := s.RefreshState(); err != nil {
+// statemgr.Writer impl.
-		return nil, fmt.Errorf("failed to load state: %w", err)
+func (s *State) WriteState(state *states.State) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	// We create a deep copy of the state here, because the caller also has
+	// a reference to the given object and can potentially go on to mutate
+	// it after we return, but we want the snapshot at this point in time.
+	s.state = state.DeepCopy()
+
+	return nil
+}
+
+// statemgr.Persister impl.
+func (s *State) PersistState(schemas *terraform.Schemas) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if s.readState != nil {
+		lineageUnchanged := s.readLineage != "" && s.lineage == s.readLineage
+		serialUnchanged := s.readSerial != 0 && s.serial == s.readSerial
+		stateUnchanged := statefile.StatesMarshalEqual(s.state, s.readState)
+		if stateUnchanged && lineageUnchanged && serialUnchanged {
+			// If the state, lineage or serial haven't changed at all then we have nothing to do.
+			return nil
+		}
+		s.serial++
+	} else {
+		// We might be writing a new state altogether, but before we do that
+		// we'll check to make sure there isn't already a snapshot present
+		// that we ought to be updating.
+		err := s.refreshState()
+		if err != nil {
+			return fmt.Errorf("failed checking for existing remote state: %s", err)
+		}
+		if s.lineage == "" { // indicates that no state snapshot is present yet
+			lineage, err := uuid.GenerateUUID()
+			if err != nil {
+				return fmt.Errorf("failed to generate initial lineage: %v", err)
+			}
+			s.lineage = lineage
+			s.serial = 0
+		}
 	}
 
-	state := s.State()
+	f := statefile.New(s.state, s.lineage, s.serial)
-	if state == nil {
+
-		// We know that there is supposed to be state (and this is not simply a new workspace
+	var buf bytes.Buffer
-		// without state) because the fallback is only invoked when outputs are present but
+	err := statefile.Write(f, &buf)
-		// detailed types are not available.
+	if err != nil {
-		return nil, ErrStateVersionUnauthorizedUpgradeState
+		return err
 	}
 
-	return state.RootModule().OutputValues, nil
+	var jsonState []byte
+	if schemas != nil {
+		jsonState, err = jsonstate.Marshal(f, schemas)
+		if err != nil {
+			return err
+		}
+	}
+
+	stateFile, err := statefile.Read(bytes.NewReader(buf.Bytes()))
+	if err != nil {
+		return fmt.Errorf("failed to read state: %w", err)
+	}
+
+	ov, err := jsonstate.MarshalOutputs(stateFile.State.RootModule().OutputValues)
+	if err != nil {
+		return fmt.Errorf("failed to translate outputs: %w", err)
+	}
+	jsonStateOutputs, err := json.Marshal(ov)
+	if err != nil {
+		return fmt.Errorf("failed to marshal outputs to json: %w", err)
+	}
+
+	err = s.uploadState(s.lineage, s.serial, s.forcePush, buf.Bytes(), jsonState, jsonStateOutputs)
+	if err != nil {
+		s.stateUploadErr = true
+		return fmt.Errorf("error uploading state: %w", err)
+	}
+	// After we've successfully persisted, what we just wrote is our new
+	// reference state until someone calls RefreshState again.
+	// We've potentially overwritten (via force) the state, lineage
+	// and / or serial (and serial was incremented) so we copy over all
+	// three fields so everything matches the new state and a subsequent
+	// operation would correctly detect no changes to the lineage, serial or state.
+	s.readState = s.state.DeepCopy()
+	s.readLineage = s.lineage
+	s.readSerial = s.serial
+	return nil
+}
+
+func (s *State) uploadState(lineage string, serial uint64, isForcePush bool, state, jsonState, jsonStateOutputs []byte) error {
+	ctx := context.Background()
+
+	options := tfe.StateVersionCreateOptions{
+		Lineage:          tfe.String(lineage),
+		Serial:           tfe.Int64(int64(serial)),
+		MD5:              tfe.String(fmt.Sprintf("%x", md5.Sum(state))),
+		State:            tfe.String(base64.StdEncoding.EncodeToString(state)),
+		Force:            tfe.Bool(isForcePush),
+		JSONState:        tfe.String(base64.StdEncoding.EncodeToString(jsonState)),
+		JSONStateOutputs: tfe.String(base64.StdEncoding.EncodeToString(jsonStateOutputs)),
+	}
+
+	// If we have a run ID, make sure to add it to the options
+	// so the state will be properly associated with the run.
+	runID := os.Getenv("TFE_RUN_ID")
+	if runID != "" {
+		options.Run = &tfe.Run{ID: runID}
+	}
+	// Create the new state.
+	_, err := s.tfeClient.StateVersions.Create(ctx, s.workspace.ID, options)
+	return err
+}
+
+// Lock calls the Client's Lock method if it's implemented.
+func (s *State) Lock(info *statemgr.LockInfo) (string, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if s.disableLocks {
+		return "", nil
+	}
+	ctx := context.Background()
+
+	lockErr := &statemgr.LockError{Info: s.lockInfo}
+
+	// Lock the workspace.
+	_, err := s.tfeClient.Workspaces.Lock(ctx, s.workspace.ID, tfe.WorkspaceLockOptions{
+		Reason: tfe.String("Locked by Terraform"),
+	})
+	if err != nil {
+		if err == tfe.ErrWorkspaceLocked {
+			lockErr.Info = info
+			err = fmt.Errorf("%s (lock ID: \"%s/%s\")", err, s.organization, s.workspace.Name)
+		}
+		lockErr.Err = err
+		return "", lockErr
+	}
+
+	s.lockInfo = info
+
+	return s.lockInfo.ID, nil
+}
+
+// statemgr.Refresher impl.
+func (s *State) RefreshState() error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.refreshState()
+}
+
+// refreshState is the main implementation of RefreshState, but split out so
+// that we can make internal calls to it from methods that are already holding
+// the s.mu lock.
+func (s *State) refreshState() error {
+	payload, err := s.getStatePayload()
+	if err != nil {
+		return err
+	}
+
+	// no remote state is OK
+	if payload == nil {
+		s.readState = nil
+		s.lineage = ""
+		s.serial = 0
+		return nil
+	}
+
+	stateFile, err := statefile.Read(bytes.NewReader(payload.Data))
+	if err != nil {
+		return err
+	}
+
+	s.lineage = stateFile.Lineage
+	s.serial = stateFile.Serial
+	s.state = stateFile.State
+
+	// Properties from the remote must be separate so we can
+	// track changes as lineage, serial and/or state are mutated
+	s.readLineage = stateFile.Lineage
+	s.readSerial = stateFile.Serial
+	s.readState = s.state.DeepCopy()
+	return nil
+}
+
+func (s *State) getStatePayload() (*remote.Payload, error) {
+	ctx := context.Background()
+
+	sv, err := s.tfeClient.StateVersions.ReadCurrent(ctx, s.workspace.ID)
+	if err != nil {
+		if err == tfe.ErrResourceNotFound {
+			// If no state exists, then return nil.
+			return nil, nil
+		}
+		return nil, fmt.Errorf("error retrieving state: %v", err)
+	}
+
+	state, err := s.tfeClient.StateVersions.Download(ctx, sv.DownloadURL)
+	if err != nil {
+		return nil, fmt.Errorf("error downloading state: %v", err)
+	}
+
+	// If the state is empty, then return nil.
+	if len(state) == 0 {
+		return nil, nil
+	}
+
+	// Get the MD5 checksum of the state.
+	sum := md5.Sum(state)
+
+	return &remote.Payload{
+		Data: state,
+		MD5:  sum[:],
+	}, nil
+}
+
+// Unlock calls the Client's Unlock method if it's implemented.
+func (s *State) Unlock(id string) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if s.disableLocks {
+		return nil
+	}
+
+	ctx := context.Background()
+
+	// We first check if there was an error while uploading the latest
+	// state. If so, we will not unlock the workspace to prevent any
+	// changes from being applied until the correct state is uploaded.
+	if s.stateUploadErr {
+		return nil
+	}
+
+	lockErr := &statemgr.LockError{Info: s.lockInfo}
+
+	// With lock info this should be treated as a normal unlock.
+	if s.lockInfo != nil {
+		// Verify the expected lock ID.
+		if s.lockInfo.ID != id {
+			lockErr.Err = fmt.Errorf("lock ID does not match existing lock")
+			return lockErr
+		}
+
+		// Unlock the workspace.
+		_, err := s.tfeClient.Workspaces.Unlock(ctx, s.workspace.ID)
+		if err != nil {
+			lockErr.Err = err
+			return lockErr
+		}
+
+		return nil
+	}
+
+	// Verify the optional force-unlock lock ID.
+	if s.organization+"/"+s.workspace.Name != id {
+		lockErr.Err = fmt.Errorf(
+			"lock ID %q does not match existing lock ID \"%s/%s\"",
+			id,
+			s.organization,
+			s.workspace.Name,
+		)
+		return lockErr
+	}
+
+	// Force unlock the workspace.
+	_, err := s.tfeClient.Workspaces.ForceUnlock(ctx, s.workspace.ID)
+	if err != nil {
+		lockErr.Err = err
+		return lockErr
+	}
+
+	return nil
+}
+
+// Delete the remote state.
+func (s *State) Delete() error {
+	err := s.tfeClient.Workspaces.Delete(context.Background(), s.organization, s.workspace.Name)
+	if err != nil && err != tfe.ErrResourceNotFound {
+		return fmt.Errorf("error deleting workspace %s: %v", s.workspace.Name, err)
+	}
+
+	return nil
+}
+
+// EnableForcePush to allow the remote client to overwrite state
+// by implementing remote.ClientForcePusher
+func (s *State) EnableForcePush() {
+	s.forcePush = true
 }
 
 // GetRootOutputValues fetches output values from Terraform Cloud
 func (s *State) GetRootOutputValues() (map[string]*states.OutputValue, error) {
 	ctx := context.Background()
 
-	so, err := s.Client.client.StateVersionOutputs.ReadCurrent(ctx, s.Client.workspace.ID)
+	so, err := s.tfeClient.StateVersionOutputs.ReadCurrent(ctx, s.workspace.ID)
 
 	if err != nil {
 		return nil, fmt.Errorf("could not read state version outputs: %w", err)
@@ -111,13 +440,27 @@ func (s *State) GetRootOutputValues() (map[string]*states.OutputValue, error) {
 			// with a version of terraform < 1.3.0. In this case, we'll eject completely from this
 			// function and fall back to the old behavior of reading the entire state file, which
 			// requires a higher level of authorization.
-			return s.fallbackReadOutputsFromFullState()
+			log.Printf("[DEBUG] falling back to reading full state")
+
+			if err := s.RefreshState(); err != nil {
+				return nil, fmt.Errorf("failed to load state: %w", err)
+			}
+
+			state := s.State()
+			if state == nil {
+				// We know that there is supposed to be state (and this is not simply a new workspace
+				// without state) because the fallback is only invoked when outputs are present but
+				// detailed types are not available.
+				return nil, ErrStateVersionUnauthorizedUpgradeState
+			}
+
+			return state.RootModule().OutputValues, nil
 		}
 
 		if output.Sensitive {
 			// Since this is a sensitive value, the output must be requested explicitly in order to
 			// read its value, which is assumed to be present by callers
-			sensitiveOutput, err := s.Client.client.StateVersionOutputs.Read(ctx, output.ID)
+			sensitiveOutput, err := s.tfeClient.StateVersionOutputs.Read(ctx, output.ID)
 			if err != nil {
 				return nil, fmt.Errorf("could not read state version output %s: %w", output.ID, err)
 			}

internal/cloud/state_test.go

@@ -1,6 +1,9 @@
 package cloud
 
 import (
+	"bytes"
+	"github.com/hashicorp/terraform/internal/states/statefile"
+	"io/ioutil"
 	"testing"
 
 	"github.com/hashicorp/go-tfe"
@@ -27,14 +30,9 @@ func TestState_GetRootOutputValues(t *testing.T) {
 	b, bCleanup := testBackendWithOutputs(t)
 	defer bCleanup()
 
-	client := &remoteClient{
+	state := &State{tfeClient: b.client, organization: b.organization, workspace: &tfe.Workspace{
-		client: b.client,
+		ID: "ws-abcd",
-		workspace: &tfe.Workspace{
+	}}
-			ID: "ws-abcd",
-		},
-	}
-
-	state := NewState(client)
 	outputs, err := state.GetRootOutputValues()
 
 	if err != nil {
@@ -81,3 +79,59 @@ func TestState_GetRootOutputValues(t *testing.T) {
 		}
 	}
 }
+
+func TestState(t *testing.T) {
+	var buf bytes.Buffer
+	s := statemgr.TestFullInitialState()
+	sf := statefile.New(s, "stub-lineage", 2)
+	err := statefile.Write(sf, &buf)
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+	data := buf.Bytes()
+
+	state := testCloudState(t)
+
+	jsonState, err := ioutil.ReadFile("../command/testdata/show-json-state/sensitive-variables/output.json")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	jsonStateOutputs := []byte(`
+{
+	"version": 4,
+	"terraform_version": "1.3.0",
+	"serial": 1,
+	"lineage": "backend-change",
+	"outputs": {
+			"foo": {
+					"type": "string",
+					"value": "bar"
+			}
+	}
+}`)
+
+	if err := state.uploadState(state.lineage, state.serial, state.forcePush, data, jsonState, jsonStateOutputs); err != nil {
+		t.Fatalf("put: %s", err)
+	}
+
+	payload, err := state.getStatePayload()
+	if err != nil {
+		t.Fatalf("get: %s", err)
+	}
+	if !bytes.Equal(payload.Data, data) {
+		t.Fatalf("expected full state %q\n\ngot: %q", string(payload.Data), string(data))
+	}
+
+	if err := state.Delete(); err != nil {
+		t.Fatalf("delete: %s", err)
+	}
+
+	p, err := state.getStatePayload()
+	if err != nil {
+		t.Fatalf("get: %s", err)
+	}
+	if p != nil {
+		t.Fatalf("expected empty state, got: %q", string(p.Data))
+	}
+}

internal/cloud/testing.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"github.com/hashicorp/terraform/internal/states/statemgr"
 	"io"
 	"net/http"
 	"net/http/httptest"
@@ -23,7 +24,6 @@ import (
 	"github.com/hashicorp/terraform/internal/configs/configschema"
 	"github.com/hashicorp/terraform/internal/httpclient"
 	"github.com/hashicorp/terraform/internal/providers"
-	"github.com/hashicorp/terraform/internal/states/remote"
 	"github.com/hashicorp/terraform/internal/terraform"
 	"github.com/hashicorp/terraform/internal/tfdiags"
 	"github.com/hashicorp/terraform/version"
@@ -110,7 +110,7 @@ func testBackendNoOperations(t *testing.T) (*Cloud, func()) {
 	return testBackend(t, obj)
 }
 
-func testRemoteClient(t *testing.T) remote.Client {
+func testCloudState(t *testing.T) *State {
 	b, bCleanup := testBackendWithName(t)
 	defer bCleanup()
 
@@ -119,7 +119,7 @@ func testRemoteClient(t *testing.T) remote.Client {
 		t.Fatalf("error: %v", err)
 	}
 
-	return raw.(*State).Client
+	return raw.(*State)
 }
 
 func testBackendWithOutputs(t *testing.T) (*Cloud, func()) {
@@ -443,3 +443,54 @@ func testVariables(s terraform.ValueSourceType, vs ...string) map[string]backend
 	}
 	return vars
 }
+
+func TestCloudLocks(t *testing.T, a, b statemgr.Full) {
+	lockerA, ok := a.(statemgr.Locker)
+	if !ok {
+		t.Fatal("client A not a statemgr.Locker")
+	}
+
+	lockerB, ok := b.(statemgr.Locker)
+	if !ok {
+		t.Fatal("client B not a statemgr.Locker")
+	}
+
+	infoA := statemgr.NewLockInfo()
+	infoA.Operation = "test"
+	infoA.Who = "clientA"
+
+	infoB := statemgr.NewLockInfo()
+	infoB.Operation = "test"
+	infoB.Who = "clientB"
+
+	lockIDA, err := lockerA.Lock(infoA)
+	if err != nil {
+		t.Fatal("unable to get initial lock:", err)
+	}
+
+	_, err = lockerB.Lock(infoB)
+	if err == nil {
+		lockerA.Unlock(lockIDA)
+		t.Fatal("client B obtained lock while held by client A")
+	}
+	if _, ok := err.(*statemgr.LockError); !ok {
+		t.Errorf("expected a LockError, but was %t: %s", err, err)
+	}
+
+	if err := lockerA.Unlock(lockIDA); err != nil {
+		t.Fatal("error unlocking client A", err)
+	}
+
+	lockIDB, err := lockerB.Lock(infoB)
+	if err != nil {
+		t.Fatal("unable to obtain lock from client B")
+	}
+
+	if lockIDB == lockIDA {
+		t.Fatalf("duplicate lock IDs: %q", lockIDB)
+	}
+
+	if err = lockerB.Unlock(lockIDB); err != nil {
+		t.Fatal("error unlocking client B:", err)
+	}
+}