Add the initial implementation for s3 locking (#2521)

Signed-off-by: yottta <andrei.ciobanu@opentofu.org>

CHANGELOG.md

@@ -11,6 +11,7 @@ NEW FEATURES:
     - `provider::terraform::decode_tfvars` - Decode a TFVars file content into an object.
     - `provider::terraform::encode_tfvars` - Encode an object into a string with the same format as a TFVars file.
     - `provider::terraform::encode_expr` - Encode an arbitrary expression into a string with valid OpenTofu syntax.
+- Added support for S3 native locking ([#599](https://github.com/opentofu/opentofu/issues/599))
 
 ENHANCEMENTS:
 * OpenTofu will now recommend using `-exclude` instead of `-target`, when possible, in the error messages about unknown values in `count` and `for_each` arguments, thereby providing a more definitive workaround. ([#2154](https://github.com/opentofu/opentofu/pull/2154))

internal/backend/remote-state/http/client.go

@@ -149,7 +149,7 @@ func (c *httpClient) Unlock(id string) error {
 	// force unlock command does not instantiate statemgr.LockInfo
 	// which means that c.jsonLockInfo will be nil
 	if c.jsonLockInfo != nil {
-		if err := json.Unmarshal(c.jsonLockInfo, &lockInfo); err != nil { //nolint:musttag // for now add musttag until we fully adopt the linting rules
+		if err := json.Unmarshal(c.jsonLockInfo, &lockInfo); err != nil {
 			return fmt.Errorf("failed to unmarshal jsonLockInfo: %w", err)
 		}
 		if lockInfo.ID != id {

internal/backend/remote-state/s3/backend.go

@@ -53,6 +53,7 @@ type Backend struct {
 	ddbTable              string
 	workspaceKeyPrefix    string
 	skipS3Checksum        bool
+	useLockfile           bool
 }
 
 // ConfigSchema returns a description of the expected configuration
@@ -453,6 +454,11 @@ See details: https://cs.opensource.google/go/x/net/+/refs/tags/v0.17.0:http/http
 				Optional:    true,
 				Description: "Do not include checksum when uploading S3 Objects. Useful for some S3-Compatible APIs as some of them do not support checksum checks.",
 			},
+			"use_lockfile": {
+				Type:        cty.Bool,
+				Optional:    true,
+				Description: "Manage locking in the same configured S3 bucket",
+			},
 		},
 	}
 }
@@ -671,6 +677,7 @@ func (b *Backend) Configure(obj cty.Value) tfdiags.Diagnostics {
 	b.serverSideEncryption = boolAttr(obj, "encrypt")
 	b.kmsKeyID = stringAttr(obj, "kms_key_id")
 	b.ddbTable = stringAttr(obj, "dynamodb_table")
+	b.useLockfile = boolAttr(obj, "use_lockfile")
 	b.skipS3Checksum = boolAttr(obj, "skip_s3_checksum")
 
 	if customerKey, ok := stringAttrOk(obj, "sse_customer_key"); ok {

internal/backend/remote-state/s3/backend_state.go

@@ -145,6 +145,7 @@ func (b *Backend) remoteClient(name string) (*RemoteClient, error) {
 		kmsKeyID:              b.kmsKeyID,
 		ddbTable:              b.ddbTable,
 		skipS3Checksum:        b.skipS3Checksum,
+		useLockfile:           b.useLockfile,
 	}
 
 	return client, nil

internal/backend/remote-state/s3/backend_test.go

@@ -1713,6 +1713,19 @@ func deleteDynamoDBTable(ctx context.Context, t *testing.T, dynClient *dynamodb.
 	}
 }
 
+func deleteDynamoEntry(ctx context.Context, t *testing.T, dynClient *dynamodb.Client, tableName string, lockId string) {
+	params := &dynamodb.DeleteItemInput{
+		Key: map[string]dtypes.AttributeValue{
+			"LockID": &dtypes.AttributeValueMemberS{Value: lockId},
+		},
+		TableName: aws.String(tableName),
+	}
+	_, err := dynClient.DeleteItem(ctx, params)
+	if err != nil {
+		t.Logf("WARNING: Failed to delete DynamoDB item %q from table %q. (error was %s)", lockId, tableName, err)
+	}
+}
+
 func populateSchema(t *testing.T, schema *configschema.Block, value cty.Value) cty.Value {
 	ty := schema.ImpliedType()
 	var path cty.Path
@@ -1810,6 +1823,14 @@ func unmarshalObject(dec cty.Value, atys map[string]cty.Type, path cty.Path) (ct
 	return cty.ObjectVal(vals), nil
 }
 
+func numberOfObjectsInBucket(t *testing.T, ctx context.Context, s3Client *s3.Client, bucketName string) int {
+	resp, err := s3Client.ListObjects(ctx, &s3.ListObjectsInput{Bucket: &bucketName})
+	if err != nil {
+		t.Fatalf("error getting objects from bucket %s: %v", bucketName, err)
+	}
+	return len(resp.Contents)
+}
+
 func must[T any](v T, err error) T {
 	if err != nil {
 		panic(err)

internal/backend/remote-state/s3/client.go

@@ -35,7 +35,10 @@ import (
 const (
 	s3EncryptionAlgorithm  = "AES256"
 	stateIDSuffix          = "-md5"
+	lockFileSuffix         = ".tflock"
 	s3ErrCodeInternalError = "InternalError"
+
+	contentTypeJSON = "application/json"
 )
 
 type RemoteClient struct {
@@ -50,6 +53,8 @@ type RemoteClient struct {
 	ddbTable              string
 
 	skipS3Checksum bool
+
+	useLockfile bool
 }
 
 var (
@@ -190,11 +195,10 @@ func (c *RemoteClient) get(ctx context.Context) (*remote.Payload, error) {
 }
 
 func (c *RemoteClient) Put(data []byte) error {
-	contentType := "application/json"
 	contentLength := int64(len(data))
 
 	i := &s3.PutObjectInput{
-		ContentType:   &contentType,
+		ContentType:   aws.String(contentTypeJSON),
 		ContentLength: aws.Int64(contentLength),
 		Body:          bytes.NewReader(data),
 		Bucket:        &c.bucketName,
@@ -272,12 +276,9 @@ func (c *RemoteClient) Delete() error {
 }
 
 func (c *RemoteClient) Lock(info *statemgr.LockInfo) (string, error) {
-	if c.ddbTable == "" {
+	if !c.IsLockingEnabled() {
 		return "", nil
 	}
-
-	info.Path = c.lockPath()
-
 	if info.ID == "" {
 		lockID, err := uuid.GenerateUUID()
 		if err != nil {
@@ -286,6 +287,26 @@ func (c *RemoteClient) Lock(info *statemgr.LockInfo) (string, error) {
 
 		info.ID = lockID
 	}
+	info.Path = c.lockPath()
+
+	if err := c.s3Lock(info); err != nil {
+		return "", err
+	}
+	if err := c.dynamoDBLock(info); err != nil {
+		// when the second lock fails from getting acquired, release the initially acquired one
+		if uErr := c.s3Unlock(info.ID); uErr != nil {
+			log.Printf("[WARN] failed to release the S3 lock after failed to acquire the dynamoDD lock: %v", uErr)
+		}
+		return "", err
+	}
+	return info.ID, nil
+}
+
+// dynamoDBLock expects the statemgr.LockInfo#ID to be filled already
+func (c *RemoteClient) dynamoDBLock(info *statemgr.LockInfo) error {
+	if c.ddbTable == "" {
+		return nil
+	}
 
 	putParams := &dynamodb.PutItemInput{
 		Item: map[string]dtypes.AttributeValue{
@@ -299,7 +320,7 @@ func (c *RemoteClient) Lock(info *statemgr.LockInfo) (string, error) {
 	ctx := context.TODO()
 	_, err := c.dynClient.PutItem(ctx, putParams)
 	if err != nil {
-		lockInfo, infoErr := c.getLockInfo(ctx)
+		lockInfo, infoErr := c.getLockInfoFromDynamoDB(ctx)
 		if infoErr != nil {
 			err = multierror.Append(err, infoErr)
 		}
@@ -308,10 +329,45 @@ func (c *RemoteClient) Lock(info *statemgr.LockInfo) (string, error) {
 			Err:  err,
 			Info: lockInfo,
 		}
-		return "", lockErr
+		return lockErr
 	}
 
-	return info.ID, nil
+	return nil
+}
+
+// s3Lock expects the statemgr.LockInfo#ID to be filled already
+func (c *RemoteClient) s3Lock(info *statemgr.LockInfo) error {
+	if !c.useLockfile {
+		return nil
+	}
+
+	lInfo := info.Marshal()
+	putParams := &s3.PutObjectInput{
+		ContentType:   aws.String(contentTypeJSON),
+		ContentLength: aws.Int64(int64(len(lInfo))),
+		Bucket:        aws.String(c.bucketName),
+		Key:           aws.String(c.lockFilePath()),
+		Body:          bytes.NewReader(lInfo),
+		IfNoneMatch:   aws.String("*"),
+	}
+
+	ctx := context.TODO()
+	ctx, _ = attachLoggerToContext(ctx)
+	_, err := c.s3Client.PutObject(ctx, putParams)
+	if err != nil {
+		lockInfo, infoErr := c.getLockInfoFromS3(ctx)
+		if infoErr != nil {
+			err = multierror.Append(err, infoErr)
+		}
+
+		lockErr := &statemgr.LockError{
+			Err:  err,
+			Info: lockInfo,
+		}
+		return lockErr
+	}
+
+	return nil
 }
 
 func (c *RemoteClient) getMD5(ctx context.Context) ([]byte, error) {
@@ -391,7 +447,7 @@ func (c *RemoteClient) deleteMD5(ctx context.Context) error {
 	return nil
 }
 
-func (c *RemoteClient) getLockInfo(ctx context.Context) (*statemgr.LockInfo, error) {
+func (c *RemoteClient) getLockInfoFromDynamoDB(ctx context.Context) (*statemgr.LockInfo, error) {
 	getParams := &dynamodb.GetItemInput{
 		Key: map[string]dtypes.AttributeValue{
 			"LockID": &dtypes.AttributeValueMemberS{Value: c.lockPath()},
@@ -426,7 +482,89 @@ func (c *RemoteClient) getLockInfo(ctx context.Context) (*statemgr.LockInfo, err
 	return lockInfo, nil
 }
 
+func (c *RemoteClient) getLockInfoFromS3(ctx context.Context) (*statemgr.LockInfo, error) {
+	getParams := &s3.GetObjectInput{
+		Bucket: aws.String(c.bucketName),
+		Key:    aws.String(c.lockFilePath()),
+	}
+
+	resp, err := c.s3Client.GetObject(ctx, getParams)
+	if err != nil {
+		var nb *types.NoSuchBucket
+		if errors.As(err, &nb) {
+			//nolint:stylecheck // error message already used in multiple places. Not recommended to be updated
+			return nil, fmt.Errorf(errS3NoSuchBucket, err)
+		}
+
+		return nil, err
+	}
+
+	lockInfo := &statemgr.LockInfo{}
+	err = json.NewDecoder(resp.Body).Decode(lockInfo)
+	if err != nil {
+		return nil, fmt.Errorf("unable to json parse the lock info %q from bucket %q: %w", c.lockFilePath(), c.bucketName, err)
+	}
+
+	return lockInfo, nil
+}
+
 func (c *RemoteClient) Unlock(id string) error {
+	// Attempt to release the lock from both sources.
+	// We want to do so to be sure that we are leaving no locks unhandled
+	s3Err := c.s3Unlock(id)
+	dynamoDBErr := c.dynamoDBUnlock(id)
+	switch {
+	case s3Err != nil && dynamoDBErr != nil:
+		s3Err.Err = multierror.Append(s3Err.Err, dynamoDBErr.Err)
+		return s3Err
+	case s3Err != nil:
+		if c.ddbTable != "" {
+			return fmt.Errorf("dynamoDB lock released but s3 failed: %w", s3Err)
+		}
+		return s3Err
+	case dynamoDBErr != nil:
+		if c.useLockfile {
+			return fmt.Errorf("s3 lock released but dynamoDB failed: %w", dynamoDBErr)
+		}
+		return dynamoDBErr
+	}
+	return nil
+}
+
+func (c *RemoteClient) s3Unlock(id string) *statemgr.LockError {
+	if !c.useLockfile {
+		return nil
+	}
+	lockErr := &statemgr.LockError{}
+	ctx := context.TODO()
+	ctx, _ = attachLoggerToContext(ctx)
+
+	lockInfo, err := c.getLockInfoFromS3(ctx)
+	if err != nil {
+		lockErr.Err = fmt.Errorf("failed to retrieve s3 lock info: %w", err)
+		return lockErr
+	}
+	lockErr.Info = lockInfo
+
+	if lockInfo.ID != id {
+		lockErr.Err = fmt.Errorf("lock id %q from s3 does not match existing lock", id)
+		return lockErr
+	}
+
+	params := &s3.DeleteObjectInput{
+		Bucket: aws.String(c.bucketName),
+		Key:    aws.String(c.lockFilePath()),
+	}
+
+	_, err = c.s3Client.DeleteObject(ctx, params)
+	if err != nil {
+		lockErr.Err = err
+		return lockErr
+	}
+	return nil
+}
+
+func (c *RemoteClient) dynamoDBUnlock(id string) *statemgr.LockError {
 	if c.ddbTable == "" {
 		return nil
 	}
@@ -434,7 +572,7 @@ func (c *RemoteClient) Unlock(id string) error {
 	lockErr := &statemgr.LockError{}
 	ctx := context.TODO()
 
-	lockInfo, err := c.getLockInfo(ctx)
+	lockInfo, err := c.getLockInfoFromDynamoDB(ctx)
 	if err != nil {
 		lockErr.Err = fmt.Errorf("failed to retrieve lock info: %w", err)
 		return lockErr
@@ -476,7 +614,11 @@ func (c *RemoteClient) getSSECustomerKeyMD5() string {
 }
 
 func (c *RemoteClient) IsLockingEnabled() bool {
-	return c.ddbTable != ""
+	return c.ddbTable != "" || c.useLockfile
+}
+
+func (c *RemoteClient) lockFilePath() string {
+	return fmt.Sprintf("%s%s", c.path, lockFileSuffix)
 }
 
 const errBadChecksumFmt = `state data in S3 does not have the expected content.

internal/backend/remote-state/s3/client_test.go

@@ -87,6 +87,126 @@ func TestRemoteClientLocks(t *testing.T) {
 	remote.TestRemoteLocks(t, s1.(*remote.State).Client, s2.(*remote.State).Client)
 }
 
+func TestRemoteS3ClientLocks(t *testing.T) {
+	testACC(t)
+	bucketName := fmt.Sprintf("%s-%x", testBucketPrefix, time.Now().Unix())
+	keyName := "testState"
+
+	b1, _ := backend.TestBackendConfig(t, New(encryption.StateEncryptionDisabled()), backend.TestWrapConfig(map[string]interface{}{
+		"bucket":       bucketName,
+		"key":          keyName,
+		"encrypt":      true,
+		"use_lockfile": true,
+	})).(*Backend)
+
+	b2, _ := backend.TestBackendConfig(t, New(encryption.StateEncryptionDisabled()), backend.TestWrapConfig(map[string]interface{}{
+		"bucket":       bucketName,
+		"key":          keyName,
+		"encrypt":      true,
+		"use_lockfile": true,
+	})).(*Backend)
+
+	ctx := context.TODO()
+	createS3Bucket(ctx, t, b1.s3Client, bucketName, b1.awsConfig.Region)
+	defer deleteS3Bucket(ctx, t, b1.s3Client, bucketName)
+
+	s1, err := b1.StateMgr(backend.DefaultStateName)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	s2, err := b2.StateMgr(backend.DefaultStateName)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	//nolint:errcheck // don't need to check the error from type assertion
+	remote.TestRemoteLocks(t, s1.(*remote.State).Client, s2.(*remote.State).Client)
+}
+
+func TestRemoteS3AndDynamoDBClientLocks(t *testing.T) {
+	testACC(t)
+	bucketName := fmt.Sprintf("%s-%x", testBucketPrefix, time.Now().Unix())
+	keyName := "testState"
+
+	b1, _ := backend.TestBackendConfig(t, New(encryption.StateEncryptionDisabled()), backend.TestWrapConfig(map[string]interface{}{
+		"bucket":         bucketName,
+		"key":            keyName,
+		"dynamodb_table": bucketName,
+		"encrypt":        true,
+	})).(*Backend)
+
+	b2, _ := backend.TestBackendConfig(t, New(encryption.StateEncryptionDisabled()), backend.TestWrapConfig(map[string]interface{}{
+		"bucket":         bucketName,
+		"key":            keyName,
+		"dynamodb_table": bucketName,
+		"encrypt":        true,
+		"use_lockfile":   true,
+	})).(*Backend)
+
+	ctx := context.TODO()
+	createS3Bucket(ctx, t, b1.s3Client, bucketName, b1.awsConfig.Region)
+	defer deleteS3Bucket(ctx, t, b1.s3Client, bucketName)
+	createDynamoDBTable(ctx, t, b1.dynClient, bucketName)
+	defer deleteDynamoDBTable(ctx, t, b1.dynClient, bucketName)
+
+	s1, err := b1.StateMgr(backend.DefaultStateName)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	s2, err := b2.StateMgr(backend.DefaultStateName)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	t.Run("dynamo lock goes first and s3+dynamo locks second", func(t *testing.T) {
+		//nolint:errcheck // don't need to check the error from type assertion
+		remote.TestRemoteLocks(t, s1.(*remote.State).Client, s2.(*remote.State).Client)
+	})
+
+	t.Run("s3+dynamo lock goes first and dynamo locks second", func(t *testing.T) {
+		//nolint:errcheck // don't need to check the error from type assertion
+		remote.TestRemoteLocks(t, s2.(*remote.State).Client, s1.(*remote.State).Client)
+	})
+}
+
+func TestRemoteS3AndDynamoDBClientLocksWithNoDBInstance(t *testing.T) {
+	testACC(t)
+	bucketName := fmt.Sprintf("%s-%x", testBucketPrefix, time.Now().Unix())
+	keyName := "testState"
+
+	b1, _ := backend.TestBackendConfig(t, New(encryption.StateEncryptionDisabled()), backend.TestWrapConfig(map[string]interface{}{
+		"bucket":         bucketName,
+		"key":            keyName,
+		"dynamodb_table": bucketName,
+		"encrypt":        true,
+		"use_lockfile":   true,
+	})).(*Backend)
+
+	ctx := context.TODO()
+	createS3Bucket(ctx, t, b1.s3Client, bucketName, b1.awsConfig.Region)
+	defer deleteS3Bucket(ctx, t, b1.s3Client, bucketName)
+
+	s1, err := b1.StateMgr(backend.DefaultStateName)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	infoA := statemgr.NewLockInfo()
+	infoA.Operation = "test"
+	infoA.Who = "clientA"
+
+	if _, err := s1.Lock(infoA); err == nil {
+		t.Fatal("unexpected successful lock: ", err)
+	}
+
+	expected := 0
+	if actual := numberOfObjectsInBucket(t, ctx, b1.s3Client, bucketName); actual != expected {
+		t.Fatalf("expected to have %d objects but got %d", expected, actual)
+	}
+}
+
 // verify that we can unlock a state with an existing lock
 func TestForceUnlock(t *testing.T) {
 	testACC(t)
@@ -180,6 +300,260 @@ func TestForceUnlock(t *testing.T) {
 	}
 }
 
+// verify that we can unlock a state with an existing lock
+func TestForceUnlockS3Only(t *testing.T) {
+	testACC(t)
+	bucketName := fmt.Sprintf("%s-force-s3-%x", testBucketPrefix, time.Now().Unix())
+	keyName := "testState"
+
+	b1, _ := backend.TestBackendConfig(t, New(encryption.StateEncryptionDisabled()), backend.TestWrapConfig(map[string]interface{}{
+		"bucket":       bucketName,
+		"key":          keyName,
+		"encrypt":      true,
+		"use_lockfile": true,
+	})).(*Backend)
+
+	b2, _ := backend.TestBackendConfig(t, New(encryption.StateEncryptionDisabled()), backend.TestWrapConfig(map[string]interface{}{
+		"bucket":       bucketName,
+		"key":          keyName,
+		"encrypt":      true,
+		"use_lockfile": true,
+	})).(*Backend)
+
+	ctx := context.TODO()
+	createS3Bucket(ctx, t, b1.s3Client, bucketName, b1.awsConfig.Region)
+	defer deleteS3Bucket(ctx, t, b1.s3Client, bucketName)
+
+	// first test with default
+	s1, err := b1.StateMgr(backend.DefaultStateName)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	info := statemgr.NewLockInfo()
+	info.Operation = "test"
+	info.Who = "clientA"
+
+	lockID, err := s1.Lock(info)
+	if err != nil {
+		t.Fatal("unable to get initial lock:", err)
+	}
+
+	// s1 is now locked, get the same state through s2 and unlock it
+	s2, err := b2.StateMgr(backend.DefaultStateName)
+	if err != nil {
+		t.Fatal("failed to get default state to force unlock:", err)
+	}
+
+	if err = s2.Unlock(lockID); err != nil {
+		t.Fatal("failed to force-unlock default state")
+	}
+
+	// now try the same thing with a named state
+	// first test with default
+	s1, err = b1.StateMgr("test")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	info = statemgr.NewLockInfo()
+	info.Operation = "test"
+	info.Who = "clientA"
+
+	lockID, err = s1.Lock(info)
+	if err != nil {
+		t.Fatal("unable to get initial lock:", err)
+	}
+
+	// s1 is now locked, get the same state through s2 and unlock it
+	s2, err = b2.StateMgr("test")
+	if err != nil {
+		t.Fatal("failed to get named state to force unlock:", err)
+	}
+
+	if err = s2.Unlock(lockID); err != nil {
+		t.Fatal("failed to force-unlock named state")
+	}
+
+	// No State lock information found for the new workspace. The client should throw the appropriate error message.
+	secondWorkspace := "new-workspace"
+	s2, err = b2.StateMgr(secondWorkspace)
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = s2.Unlock(lockID)
+	if err == nil {
+		t.Fatal("expected an error to occur:", err)
+	}
+	expectedErrorMsg := fmt.Errorf("failed to retrieve s3 lock info: operation error S3: GetObject, https response error StatusCode: 404")
+	if !strings.HasPrefix(err.Error(), expectedErrorMsg.Error()) {
+		t.Errorf("Unlock()\nactual = %v\nexpected = %v", err, expectedErrorMsg)
+	}
+}
+
+// verify that we can unlock a state with an existing lock
+func TestForceUnlockS3AndDynamo(t *testing.T) {
+	testACC(t)
+	bucketName := fmt.Sprintf("%s-force-s3-dynamo-%x", testBucketPrefix, time.Now().Unix())
+	keyName := "testState"
+
+	b1, _ := backend.TestBackendConfig(t, New(encryption.StateEncryptionDisabled()), backend.TestWrapConfig(map[string]interface{}{
+		"bucket":         bucketName,
+		"key":            keyName,
+		"encrypt":        true,
+		"use_lockfile":   true,
+		"dynamodb_table": bucketName,
+	})).(*Backend)
+
+	b2, _ := backend.TestBackendConfig(t, New(encryption.StateEncryptionDisabled()), backend.TestWrapConfig(map[string]interface{}{
+		"bucket":         bucketName,
+		"key":            keyName,
+		"encrypt":        true,
+		"use_lockfile":   true,
+		"dynamodb_table": bucketName,
+	})).(*Backend)
+
+	ctx := context.TODO()
+	createS3Bucket(ctx, t, b1.s3Client, bucketName, b1.awsConfig.Region)
+	defer deleteS3Bucket(ctx, t, b1.s3Client, bucketName)
+	createDynamoDBTable(ctx, t, b1.dynClient, bucketName)
+	defer deleteDynamoDBTable(ctx, t, b1.dynClient, bucketName)
+
+	// first test with default
+	s1, err := b1.StateMgr(backend.DefaultStateName)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	info := statemgr.NewLockInfo()
+	info.Operation = "test"
+	info.Who = "clientA"
+
+	lockID, err := s1.Lock(info)
+	if err != nil {
+		t.Fatal("unable to get initial lock:", err)
+	}
+
+	// s1 is now locked, get the same state through s2 and unlock it
+	s2, err := b2.StateMgr(backend.DefaultStateName)
+	if err != nil {
+		t.Fatal("failed to get default state to force unlock:", err)
+	}
+
+	if err = s2.Unlock(lockID); err != nil {
+		t.Fatal("failed to force-unlock default state")
+	}
+
+	// now try the same thing with a named state
+	// first test with default
+	s1, err = b1.StateMgr("test")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	info = statemgr.NewLockInfo()
+	info.Operation = "test"
+	info.Who = "clientA"
+
+	lockID, err = s1.Lock(info)
+	if err != nil {
+		t.Fatal("unable to get initial lock:", err)
+	}
+
+	// s1 is now locked, get the same state through s2 and unlock it
+	s2, err = b2.StateMgr("test")
+	if err != nil {
+		t.Fatal("failed to get named state to force unlock:", err)
+	}
+
+	if err = s2.Unlock(lockID); err != nil {
+		t.Fatal("failed to force-unlock named state")
+	}
+
+	// No State lock information found for the new workspace. The client should throw the appropriate error message.
+	secondWorkspace := "new-workspace"
+	s2, err = b2.StateMgr(secondWorkspace)
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = s2.Unlock(lockID)
+	if err == nil {
+		t.Fatal("expected an error to occur:", err)
+	}
+	expectedErrorMsg := []error{
+		fmt.Errorf("failed to retrieve s3 lock info: operation error S3: GetObject, https response error StatusCode: 404"),
+		fmt.Errorf("failed to retrieve lock info: no lock info found for: \"%s/env:/%s/%s\" within the DynamoDB table: %s", bucketName, secondWorkspace, keyName, bucketName),
+	}
+	for _, expectedErr := range expectedErrorMsg {
+		if !strings.Contains(err.Error(), expectedErr.Error()) {
+			t.Errorf("Unlock() should contain expected.\nactual = %v\nexpected = %v", err, expectedErr)
+		}
+	}
+}
+
+// verify the way it's handled the situation when the lock is in S3 but not in DynamoDB
+func TestForceUnlockS3WithAndDynamoWithout(t *testing.T) {
+	testACC(t)
+	bucketName := fmt.Sprintf("%s-force-s3-dynamo-%x", testBucketPrefix, time.Now().Unix())
+	keyName := "testState"
+
+	b1, _ := backend.TestBackendConfig(t, New(encryption.StateEncryptionDisabled()), backend.TestWrapConfig(map[string]interface{}{
+		"bucket":         bucketName,
+		"key":            keyName,
+		"encrypt":        true,
+		"use_lockfile":   true,
+		"dynamodb_table": bucketName,
+	})).(*Backend)
+
+	ctx := context.TODO()
+	createS3Bucket(ctx, t, b1.s3Client, bucketName, b1.awsConfig.Region)
+	defer deleteS3Bucket(ctx, t, b1.s3Client, bucketName)
+	createDynamoDBTable(ctx, t, b1.dynClient, bucketName)
+	defer deleteDynamoDBTable(ctx, t, b1.dynClient, bucketName)
+
+	// first create both locks: s3 and dynamo
+	s1, err := b1.StateMgr(backend.DefaultStateName)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	info := statemgr.NewLockInfo()
+	info.Operation = "test"
+	info.Who = "clientA"
+
+	lockID, err := s1.Lock(info)
+	if err != nil {
+		t.Fatal("unable to get initial lock:", err)
+	}
+
+	// Remove the dynamo lock to simulate that the lock in s3 was acquired, dynamo failed but s3 release failed in the end.
+	// Therefore, the user is left in the situation with s3 lock existing and dynamo missing.
+	deleteDynamoEntry(ctx, t, b1.dynClient, bucketName, info.Path)
+	err = s1.Unlock(lockID)
+	if err == nil {
+		t.Fatal("expected to get an error but got nil")
+	}
+	expectedErrMsg := fmt.Sprintf("s3 lock released but dynamoDB failed: failed to retrieve lock info: no lock info found for: %q within the DynamoDB table: %s", info.Path, bucketName)
+	if err.Error() != expectedErrMsg {
+		t.Fatalf("unexpected error message.\nexpected: %s\nactual:%s", expectedErrMsg, err.Error())
+	}
+
+	// Now, unlocking should fail with error on both locks
+	err = s1.Unlock(lockID)
+	if err == nil {
+		t.Fatal("expected to get an error but got nil")
+	}
+	expectedErrMsgs := []string{
+		fmt.Sprintf("failed to retrieve lock info: no lock info found for: %q within the DynamoDB table: %s", info.Path, bucketName),
+		"failed to retrieve s3 lock info: operation error S3: GetObject, https response error StatusCode: 404",
+	}
+	for _, expectedErrorMsg := range expectedErrMsgs {
+		if !strings.Contains(err.Error(), expectedErrorMsg) {
+			t.Fatalf("returned error does not contain the expected content.\nexpected: %s\nactual:%s", expectedErrorMsg, err.Error())
+		}
+	}
+}
+
 func TestRemoteClient_clientMD5(t *testing.T) {
 	testACC(t)
 
@@ -349,6 +723,7 @@ func TestRemoteClient_IsLockingEnabled(t *testing.T) {
 	tests := []struct {
 		name        string
 		ddbTable    string
+		useLockfile bool
 		wantResult  bool
 	}{
 		{
@@ -361,12 +736,31 @@ func TestRemoteClient_IsLockingEnabled(t *testing.T) {
 			ddbTable:   "",
 			wantResult: false,
 		},
+		{
+			name:        "Locking disabled when ddbTable is empty and useLockfile disabled",
+			ddbTable:    "",
+			useLockfile: false,
+			wantResult:  false,
+		},
+		{
+			name:        "Locking enabled when ddbTable is set or useLockfile enabled",
+			ddbTable:    "my-lock-table",
+			useLockfile: true,
+			wantResult:  true,
+		},
+		{
+			name:        "Locking enabled when ddbTable is empty and useLockfile enabled",
+			ddbTable:    "",
+			useLockfile: true,
+			wantResult:  true,
+		},
 	}
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			client := &RemoteClient{
 				ddbTable:    tt.ddbTable,
+				useLockfile: tt.useLockfile,
 			}
 
 			gotResult := client.IsLockingEnabled()

internal/states/statemgr/locker.go

@@ -134,25 +134,25 @@ type LockInfo struct {
 	// Unique ID for the lock. NewLockInfo provides a random ID, but this may
 	// be overridden by the lock implementation. The final value of ID will be
 	// returned by the call to Lock.
-	ID string
+	ID string `json:"ID"`
 
 	// OpenTofu operation, provided by the caller.
-	Operation string
+	Operation string `json:"Operation"`
 
 	// Extra information to store with the lock, provided by the caller.
-	Info string
+	Info string `json:"Info"`
 
 	// user@hostname when available
-	Who string
+	Who string `json:"Who"`
 
 	// OpenTofu version
-	Version string
+	Version string `json:"Version"`
 
 	// Time that the lock was taken.
-	Created time.Time
+	Created time.Time `json:"Created"`
 
 	// Path to the state file when applicable. Set by the Lock implementation.
-	Path string
+	Path string `json:"Path"`
 }
 
 // NewLockInfo creates a LockInfo object and populates many of its fields

website/docs/language/settings/backends/s3.mdx

@@ -7,8 +7,9 @@ description: OpenTofu can store state remotely in S3 and lock that state with Dy
 
 Stores the state as a given key in a given bucket on
 [Amazon S3](https://aws.amazon.com/s3/).
-This backend also supports state locking and consistency checking via
-[Dynamo DB](https://aws.amazon.com/dynamodb/), which can be enabled by setting
+This backend supports multiple locking mechanisms. The preferred one is a native S3 locking via
+conditional writes with `If-None-Match` header. This can be enabled by setting `use_lockfile=true`.
+Another option is to use [Dynamo DB](https://aws.amazon.com/dynamodb/) locking, which can be enabled by setting
 the `dynamodb_table` field to an existing DynamoDB table name.
 A single DynamoDB table can be used to lock multiple remote state files. OpenTofu generates key names that include the values of the `bucket` and `key` variables.
 
@@ -18,6 +19,10 @@ It is highly recommended that you enable
 on the S3 bucket to allow for state recovery in the case of accidental deletions and human error.
 :::
 
+:::info
+For a smooth transition to the S3 locking, please read the [dedicated section](#s3-state-locking).
+:::
+
 ## Example Configuration
 
 ```hcl
@@ -352,6 +357,32 @@ The following configuration is optional:
 * `dynamodb_endpoint` - (Optional) **Deprecated** Custom endpoint for the AWS DynamoDB API. This can also be sourced from the `AWS_DYNAMODB_ENDPOINT` environment variable.
 * `dynamodb_table` - (Optional) Name of DynamoDB Table to use for state locking and consistency. The table must have a partition key named `LockID` with type of `String`. If not configured, state locking will be disabled.
 
+### S3 State Locking
+
+* `use_lockfile` - (Optional) Enable locking directly into the configured bucket for the state.
+
+To migrate from DynamoDB to S3 locking, the following steps can be followed:
+1. The new attribute `use_lockfile=true` can be added alongside `dynamodb_table`:
+   * With both attributes specified, OpenTofu will try to acquire the lock first in S3 and if successful, will try to acquire the lock in DynamoDB. In this case, the lock will be considered acquired only when both (S3 and DynamoDB) locks were acquired successfully.
+   * Later, after a baking period with both locking mechanisms enabled, if no issues encountered, remove the `dynamodb_table` attribute. Now, you are solely on the S3 locking.
+   * **Info:** Keeping both locking mechanisms enabled, ensures that nobody will acquire the lock regardless of having or not the latest configuration.
+2. The new attribute `use_lockfile=true` can be added and `dynamodb_table` removed:
+   * This will switch from DynamoDB to S3 locking. **Caution:** when the updated configuration is executed from multiple places (multiple machines, pipelines on PRs, etc), you might get into issues where one outdated copy of the configuration is using DynamoDB locking and the one updated is using S3 locking. This could end up in concurrent access on the same state file.
+   * Once the state is updated by using this approach, the state digest that OpenTofu was storing in DynamoDB (for data consistency checks) will get stale. If you wish to go back to DynamoDB locking, **the old digest needs to be cleaned up manually**.
+
+:::note
+Remember, any changes to the `backend` block will require to run `tofu init -reconfigure`.
+:::
+
+:::note
+As mentioned in the beginning of this page, OpenTofu recommends to have versioning enabled on the S3 bucket where state file(s) are stored.
+By setting `use_lockfile=true`, acquiring and releasing locks will add a good amount of writes and reads to the bucket.
+Therefore, for a versioning-enabled bucket, the number of versions for that object could grow significantly.
+Even though the cost should be negligible for the locking objects, a lifecycle configuration of the S3 bucket to limit the number of versions of an object would be advised.
+:::
+
+When it comes to the workspace usage, the S3 locking will behave normally, storing the lock file right next to its related state object.
+
 ## Multi-account AWS Architecture
 
 A common architectural pattern is for an organization to use a number of