feat: RDS tag actions, --tags on create, and the ARN parser (#595)

spinifex/awsec2query/parser.go

@@ -94,7 +94,7 @@ func setStructFields(v reflect.Value, params map[string]string, prefix string) e
 		// Handle slice fields (e.g., SecurityGroup.1, SecurityGroup.2)
 		if field.Kind() == reflect.Slice {
 			for _, queryKey := range queryKeys {
-				if err := setSliceField(field, params, queryKey); err != nil {
+				if err := setSliceField(field, params, queryKey, fieldType.Tag.Get("locationNameList")); err != nil {
 					return fmt.Errorf("error setting slice field %s: %w", fieldName, err)
 				}
 				if field.Len() > 0 {
@@ -108,7 +108,7 @@ func setStructFields(v reflect.Value, params map[string]string, prefix string) e
 		if field.Kind() == reflect.Pointer && field.Type().Elem().Kind() == reflect.Slice {
 			for _, queryKey := range queryKeys {
 				sliceField := reflect.New(field.Type().Elem()).Elem()
-				if err := setSliceField(sliceField, params, queryKey); err != nil {
+				if err := setSliceField(sliceField, params, queryKey, fieldType.Tag.Get("locationNameList")); err != nil {
 					return fmt.Errorf("error setting pointer to slice field %s: %w", fieldName, err)
 				}
 				if sliceField.Len() > 0 {
@@ -181,27 +181,42 @@ func setFieldValue(field reflect.Value, value string) error {
 // from adversarial indexes like `Filter.999999`.
 const maxSliceLen = 1024
 
-func setSliceField(field reflect.Value, params map[string]string, prefix string) error {
-	// Collect indexed items; supports EC2-style (Prefix.N) and IAM/ELBv2-style (Prefix.member.N).
-	indices := make(map[int]bool)
-	useMember := false
+// Records the 1-based indices of the entries stored under prefix, reporting
+// whether it matched anything at all.
+func collectIndices(params map[string]string, prefix string, indices map[int]bool) bool {
+	found := false
 	for key := range params {
-		if strings.HasPrefix(key, prefix+".member.") {
-			parts := strings.Split(key[len(prefix)+len(".member."):], ".")
-			if len(parts) > 0 {
-				if idx, err := strconv.Atoi(parts[0]); err == nil {
-					indices[idx] = true
-					useMember = true
-				}
-			}
-		} else if strings.HasPrefix(key, prefix+".") {
-			parts := strings.Split(key[len(prefix)+1:], ".")
-			if len(parts) > 0 {
-				if idx, err := strconv.Atoi(parts[0]); err == nil {
-					indices[idx] = true
-				}
-			}
+		if !strings.HasPrefix(key, prefix) {
+			continue
 		}
+		head, _, _ := strings.Cut(key[len(prefix):], ".")
+		if idx, err := strconv.Atoi(head); err == nil {
+			indices[idx] = true
+			found = true
+		}
+	}
+	return found
+}
+
+// listName is the shape's locationNameList, if it declares one. Empty for the
+// majority of shapes, which use the default wrapper or none at all.
+func setSliceField(field reflect.Value, params map[string]string, prefix, listName string) error {
+	// Collect indexed items. A list is serialized EC2-style (Prefix.N),
+	// IAM/ELBv2-style (Prefix.member.N), or under its own locationNameList
+	// (Prefix.Tag.N, which is how RDS writes a tag list).
+	indices := make(map[int]bool)
+	wrapper := ""
+	for _, candidate := range []string{"member", listName} {
+		if candidate == "" {
+			continue
+		}
+		if collectIndices(params, prefix+"."+candidate+".", indices) {
+			wrapper = candidate
+			break
+		}
+	}
+	if wrapper == "" {
+		collectIndices(params, prefix+".", indices)
 	}
 
 	if len(indices) == 0 {
@@ -226,11 +241,9 @@ func setSliceField(field reflect.Value, params map[string]string, prefix string)
 	// Process each index
 	for idx := 1; idx <= denseLen; idx++ {
 		elem := slice.Index(idx - 1)
-		var indexPrefix string
-		if useMember {
-			indexPrefix = fmt.Sprintf("%s.member.%d", prefix, idx)
-		} else {
-			indexPrefix = fmt.Sprintf("%s.%d", prefix, idx)
+		indexPrefix := fmt.Sprintf("%s.%d", prefix, idx)
+		if wrapper != "" {
+			indexPrefix = fmt.Sprintf("%s.%s.%d", prefix, wrapper, idx)
 		}
 
 		// Handle different element types

spinifex/awsec2query/parser_test.go

@@ -8,6 +8,7 @@ import (
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/aws/aws-sdk-go/service/elbv2"
+	"github.com/aws/aws-sdk-go/service/rds"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -475,6 +476,43 @@ func TestQueryParamsToStruct_LocationNameDryRun(t *testing.T) {
 	assert.True(t, aws.BoolValue(input.DryRun))
 }
 
+// RDS declares locationNameList:"Tag" on its tag lists, so the wrapper segment
+// is "Tag" rather than the default "member". Parsing it as either of the other
+// two shapes would drop every tag silently.
+func TestQueryParamsToStruct_LocationNameListWrapper(t *testing.T) {
+	args := map[string]string{
+		"ResourceName":     "arn:aws:rds:ap-southeast-2:123456789012:db:orders-db",
+		"Tags.Tag.1.Key":   "env",
+		"Tags.Tag.1.Value": "prod",
+		"Tags.Tag.2.Key":   "team",
+		"Tags.Tag.2.Value": "platform",
+	}
+
+	input := &rds.AddTagsToResourceInput{}
+	require.NoError(t, QueryParamsToStruct(args, input))
+
+	require.Len(t, input.Tags, 2)
+	assert.Equal(t, "env", aws.StringValue(input.Tags[0].Key))
+	assert.Equal(t, "prod", aws.StringValue(input.Tags[0].Value))
+	assert.Equal(t, "team", aws.StringValue(input.Tags[1].Key))
+	assert.Equal(t, "platform", aws.StringValue(input.Tags[1].Value))
+}
+
+// A shape that declares a locationNameList still accepts the default wrapper,
+// so an SDK that serializes the list either way is understood.
+func TestQueryParamsToStruct_LocationNameListFallsBackToMember(t *testing.T) {
+	args := map[string]string{
+		"Tags.member.1.Key":   "env",
+		"Tags.member.1.Value": "prod",
+	}
+
+	input := &rds.AddTagsToResourceInput{}
+	require.NoError(t, QueryParamsToStruct(args, input))
+
+	require.Len(t, input.Tags, 1)
+	assert.Equal(t, "env", aws.StringValue(input.Tags[0].Key))
+}
+
 func TestQueryParamsToStruct_NonStructPointer(t *testing.T) {
 	args := map[string]string{"Key": "value"}
 	str := "hello"

spinifex/daemon/daemon.go

@@ -1058,6 +1058,9 @@ func (d *Daemon) subscribeAll() error {
 			natsSub{handlers_rds.SubjectGetDBBootstrapConfig, handleNATSRequest(d.rdsService.GetDBBootstrapConfig), "spinifex-workers"},
 			natsSub{handlers_rds.SubjectCreateDBInstance, handleNATSRequest(d.rdsService.CreateDBInstance), "spinifex-workers"},
 			natsSub{handlers_rds.SubjectDescribeDBInstances, handleNATSRequest(d.rdsService.DescribeDBInstances), "spinifex-workers"},
+			natsSub{handlers_rds.SubjectAddTagsToResource, handleNATSRequest(d.rdsService.AddTagsToResource), "spinifex-workers"},
+			natsSub{handlers_rds.SubjectRemoveTagsFromResource, handleNATSRequest(d.rdsService.RemoveTagsFromResource), "spinifex-workers"},
+			natsSub{handlers_rds.SubjectListTagsForResource, handleNATSRequest(d.rdsService.ListTagsForResource), "spinifex-workers"},
 		)
 	}
 

spinifex/gateway/rds/handler.go

@@ -109,9 +109,9 @@ var actions = map[string]Handler{
 	"DeleteDBParameterGroup":    pending(),
 
 	// Tags.
-	"AddTagsToResource":      pending(),
-	"RemoveTagsFromResource": pending(),
-	"ListTagsForResource":    pending(),
+	"AddTagsToResource":      typed(AddTagsToResource),
+	"RemoveTagsFromResource": typed(RemoveTagsFromResource),
+	"ListTagsForResource":    typed(ListTagsForResource),
 
 	// Events.
 	"DescribeEvents": pending(),

spinifex/gateway/rds/handler_test.go

@@ -102,7 +102,10 @@ func TestDispatch_UnknownAction(t *testing.T) {
 
 // The customer actions this phase implements forward to the daemon, so they are
 // dispatched against a stub responder rather than a nil connection.
-var liveActions = []string{"CreateDBInstance", "DescribeDBInstances"}
+var liveActions = []string{
+	"CreateDBInstance", "DescribeDBInstances",
+	"AddTagsToResource", "RemoveTagsFromResource", "ListTagsForResource",
+}
 
 // What is under test in this file is the action table and the XML envelope, not
 // the orchestration behind the subject, so the responder returns a fixed output.
@@ -113,6 +116,12 @@ func newStubbedNATS(t *testing.T) *nats.Conn {
 		&rds.CreateDBInstanceOutput{DBInstance: &rds.DBInstance{DBInstanceIdentifier: aws.String("orders-db")}})
 	respondWith(t, nc, handlers_rds.SubjectDescribeDBInstances,
 		&rds.DescribeDBInstancesOutput{DBInstances: []*rds.DBInstance{}})
+	respondWith(t, nc, handlers_rds.SubjectAddTagsToResource, &rds.AddTagsToResourceOutput{})
+	respondWith(t, nc, handlers_rds.SubjectRemoveTagsFromResource, &rds.RemoveTagsFromResourceOutput{})
+	respondWith(t, nc, handlers_rds.SubjectListTagsForResource,
+		&rds.ListTagsForResourceOutput{TagList: []*rds.Tag{
+			{Key: aws.String("env"), Value: aws.String("prod")},
+		}})
 	return nc
 }
 
@@ -188,6 +197,90 @@ func TestDispatch_DescribeDBInstancesReturnsEmptyResultSet(t *testing.T) {
 		string(body))
 }
 
+// AWS nests each tag inside the list rather than flattening it, and the SDK's
+// unmarshaler will not find a tag any other way.
+func TestDispatch_ListTagsForResourceRendersTheNestedTagList(t *testing.T) {
+	body, err := Dispatch(t.Context(), "ListTagsForResource", map[string]string{
+		"Action":       "ListTagsForResource",
+		"ResourceName": handlers_rds.DBInstanceARN("ap-southeast-2", testAccountID, "orders-db"),
+	}, newStubbedNATS(t), testCaller)
+	require.NoError(t, err)
+
+	assert.Equal(t,
+		"<ListTagsForResourceResponse><ListTagsForResourceResult><TagList>"+
+			"<Tag><Key>env</Key><Value>prod</Value></Tag>"+
+			"</TagList></ListTagsForResourceResult></ListTagsForResourceResponse>",
+		string(body))
+}
+
+// RDS serializes a tag list under its own locationNameList, so the params
+// arrive as Tags.Tag.N.Key. Parsing them as anything else would drop the tags
+// and report a success that tagged nothing.
+func TestDispatch_AddTagsToResourceParsesTheTagList(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	requests := make(chan rds.AddTagsToResourceInput, 1)
+	sub, err := nc.Subscribe(handlers_rds.SubjectAddTagsToResource, func(msg *nats.Msg) {
+		var in rds.AddTagsToResourceInput
+		if err := json.Unmarshal(msg.Data, &in); err != nil {
+			t.Errorf("unmarshal AddTagsToResource request: %v", err)
+			return
+		}
+		requests <- in
+		if err := msg.Respond([]byte(`{}`)); err != nil {
+			t.Logf("respond on %s: %v", handlers_rds.SubjectAddTagsToResource, err)
+		}
+	})
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = sub.Unsubscribe() })
+
+	arn := handlers_rds.DBInstanceARN("ap-southeast-2", testAccountID, "orders-db")
+	_, err = Dispatch(t.Context(), "AddTagsToResource", map[string]string{
+		"Action":         "AddTagsToResource",
+		"ResourceName":   arn,
+		"Tags.Tag.1.Key": "env", "Tags.Tag.1.Value": "prod",
+		"Tags.Tag.2.Key": "team", "Tags.Tag.2.Value": "platform",
+	}, nc, testCaller)
+	require.NoError(t, err)
+
+	got := <-requests
+	assert.Equal(t, arn, aws.StringValue(got.ResourceName))
+	require.Len(t, got.Tags, 2)
+	assert.Equal(t, "env", aws.StringValue(got.Tags[0].Key))
+	assert.Equal(t, "prod", aws.StringValue(got.Tags[0].Value))
+	assert.Equal(t, "team", aws.StringValue(got.Tags[1].Key))
+	assert.Equal(t, "platform", aws.StringValue(got.Tags[1].Value))
+}
+
+// TagKeys carries no locationNameList, so it keeps the default member wrapper
+// even though the sibling tag list does not.
+func TestDispatch_RemoveTagsFromResourceParsesMemberTagKeys(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	requests := make(chan rds.RemoveTagsFromResourceInput, 1)
+	sub, err := nc.Subscribe(handlers_rds.SubjectRemoveTagsFromResource, func(msg *nats.Msg) {
+		var in rds.RemoveTagsFromResourceInput
+		if err := json.Unmarshal(msg.Data, &in); err != nil {
+			t.Errorf("unmarshal RemoveTagsFromResource request: %v", err)
+			return
+		}
+		requests <- in
+		if err := msg.Respond([]byte(`{}`)); err != nil {
+			t.Logf("respond on %s: %v", handlers_rds.SubjectRemoveTagsFromResource, err)
+		}
+	})
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = sub.Unsubscribe() })
+
+	_, err = Dispatch(t.Context(), "RemoveTagsFromResource", map[string]string{
+		"Action":           "RemoveTagsFromResource",
+		"ResourceName":     handlers_rds.DBInstanceARN("ap-southeast-2", testAccountID, "orders-db"),
+		"TagKeys.member.1": "env",
+		"TagKeys.member.2": "team",
+	}, nc, testCaller)
+	require.NoError(t, err)
+
+	assert.Equal(t, []string{"env", "team"}, aws.StringValueSlice((<-requests).TagKeys))
+}
+
 // A filtered describe is still an empty result set, not a parse failure: the
 // query params have to survive QueryParamsToStruct into the typed input.
 func TestDispatch_DescribeDBInstancesWithFilters(t *testing.T) {

spinifex/gateway/rds/tags.go

@@ -0,0 +1,23 @@
+package gateway_rds
+
+import (
+	"context"
+
+	"github.com/aws/aws-sdk-go/service/rds"
+	handlers_rds "github.com/mulgadc/spinifex/spinifex/handlers/rds"
+	"github.com/nats-io/nats.go"
+)
+
+// The ARN in the request names the resource, and the daemon validates it
+// against the caller's account — the account is never taken from the ARN.
+func AddTagsToResource(ctx context.Context, input *rds.AddTagsToResourceInput, nc *nats.Conn, caller Caller) (any, error) {
+	return handlers_rds.NewNATSService(nc).AddTagsToResource(ctx, input, caller.AccountID)
+}
+
+func RemoveTagsFromResource(ctx context.Context, input *rds.RemoveTagsFromResourceInput, nc *nats.Conn, caller Caller) (any, error) {
+	return handlers_rds.NewNATSService(nc).RemoveTagsFromResource(ctx, input, caller.AccountID)
+}
+
+func ListTagsForResource(ctx context.Context, input *rds.ListTagsForResourceInput, nc *nats.Conn, caller Caller) (any, error) {
+	return handlers_rds.NewNATSService(nc).ListTagsForResource(ctx, input, caller.AccountID)
+}

spinifex/handlers/rds/arn.go

@@ -1,6 +1,11 @@
 package handlers_rds
 
-import "fmt"
+import (
+	"fmt"
+	"strings"
+
+	"github.com/mulgadc/spinifex/spinifex/awserrors"
+)
 
 // RDS separates resource type and name with a colon, unlike the slash-separated
 // ECS/EKS ARNs, so resource-scoped IAM policies and Terraform state round-trip
@@ -22,3 +27,71 @@ func DBSubnetGroupARN(region, accountID, name string) string {
 func DBParameterGroupARN(region, accountID, name string) string {
 	return fmt.Sprintf("arn:aws:rds:%s:%s:pg:%s", region, accountID, name)
 }
+
+// The resource-type segment of an RDS ARN, and the key the tag registry and
+// resource-scoped authorization dispatch on.
+type ResourceKind string
+
+const (
+	ResourceKindDBInstance       ResourceKind = "db"
+	ResourceKindDBSnapshot       ResourceKind = "snapshot"
+	ResourceKindDBSubnetGroup    ResourceKind = "subgrp"
+	ResourceKindDBParameterGroup ResourceKind = "pg"
+)
+
+func validResourceKind(kind ResourceKind) bool {
+	switch kind {
+	case ResourceKindDBInstance, ResourceKindDBSnapshot, ResourceKindDBSubnetGroup, ResourceKindDBParameterGroup:
+		return true
+	default:
+		return false
+	}
+}
+
+// An ARN split into the parts a caller acts on. Partition and service are
+// validated rather than returned: only "arn:aws:rds" is ever accepted.
+type ParsedARN struct {
+	Region     string
+	AccountID  string
+	Kind       ResourceKind
+	Identifier string
+}
+
+// arn:aws:rds:{region}:{accountID}:{kind}:{identifier}
+const arnSegmentCount = 7
+
+// Parses one of the four D17 ARNs and validates it belongs to the caller.
+// Region and account are checked here rather than at policy evaluation, so a
+// foreign-account reference never reaches the evaluator at all.
+func ParseARN(arn, region, accountID string) (ParsedARN, error) {
+	// The identifier cannot contain a colon, so the trailing segment SplitN
+	// leaves intact is rejected below if it holds one.
+	parts := strings.SplitN(arn, ":", arnSegmentCount)
+	if len(parts) != arnSegmentCount {
+		return ParsedARN{}, arnError(arn, "expected the form arn:aws:rds:{region}:{account}:{type}:{name}")
+	}
+	if parts[0] != "arn" || parts[1] != "aws" || parts[2] != "rds" {
+		return ParsedARN{}, arnError(arn, "only arn:aws:rds resources are addressable here")
+	}
+	if parts[3] != region {
+		return ParsedARN{}, arnError(arn, fmt.Sprintf("region %q does not match this endpoint's region %q", parts[3], region))
+	}
+	if parts[4] != accountID {
+		return ParsedARN{}, arnError(arn, "the resource belongs to another account")
+	}
+
+	kind := ResourceKind(parts[5])
+	if !validResourceKind(kind) {
+		return ParsedARN{}, arnError(arn, fmt.Sprintf("%q is not an RDS resource type", parts[5]))
+	}
+	identifier := parts[6]
+	if identifier == "" || strings.ContainsAny(identifier, ":/") {
+		return ParsedARN{}, arnError(arn, "the resource name is empty or malformed")
+	}
+
+	return ParsedARN{Region: parts[3], AccountID: parts[4], Kind: kind, Identifier: identifier}, nil
+}
+
+func arnError(arn, why string) error {
+	return fmt.Errorf("%s: %q is not a valid RDS ARN: %s", awserrors.ErrorInvalidParameterValue, arn, why)
+}

spinifex/handlers/rds/arn_parse_test.go

@@ -0,0 +1,76 @@
+package handlers_rds
+
+import (
+	"testing"
+
+	"github.com/mulgadc/spinifex/spinifex/awserrors"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// Every D17 ARN the builders mint must parse back into the parts it was built
+// from, so a resource named by a describe is addressable by a tag action.
+func TestParseARN_RoundTripsEveryResourceType(t *testing.T) {
+	cases := []struct {
+		kind       ResourceKind
+		identifier string
+		arn        string
+	}{
+		{ResourceKindDBInstance, "orders-db", DBInstanceARN(testRegion, testAccountID, "orders-db")},
+		{ResourceKindDBSnapshot, "orders-db-2026-07-24", DBSnapshotARN(testRegion, testAccountID, "orders-db-2026-07-24")},
+		{ResourceKindDBSubnetGroup, "prod-db-subnets", DBSubnetGroupARN(testRegion, testAccountID, "prod-db-subnets")},
+		{ResourceKindDBParameterGroup, "postgres16-tuned", DBParameterGroupARN(testRegion, testAccountID, "postgres16-tuned")},
+	}
+
+	for _, tc := range cases {
+		t.Run(string(tc.kind), func(t *testing.T) {
+			parsed, err := ParseARN(tc.arn, testRegion, testAccountID)
+			require.NoError(t, err)
+			assert.Equal(t, tc.kind, parsed.Kind)
+			assert.Equal(t, tc.identifier, parsed.Identifier)
+			assert.Equal(t, testRegion, parsed.Region)
+			assert.Equal(t, testAccountID, parsed.AccountID)
+		})
+	}
+}
+
+func TestParseARN_RejectsMalformedAndForeignARNs(t *testing.T) {
+	cases := []struct {
+		name string
+		arn  string
+	}{
+		// The ECS shape. Copying its parser would accept this and read the whole
+		// "db/orders-db" as the resource type.
+		{"slash delimited", "arn:aws:rds:" + testRegion + ":" + testAccountID + ":db/orders-db"},
+		{"truncated", "arn:aws:rds:" + testRegion + ":" + testAccountID + ":db"},
+		{"extra segment", DBInstanceARN(testRegion, testAccountID, "orders-db") + ":extra"},
+		{"empty identifier", "arn:aws:rds:" + testRegion + ":" + testAccountID + ":db:"},
+		{"another service", "arn:aws:ec2:" + testRegion + ":" + testAccountID + ":db:orders-db"},
+		{"another partition", "arn:aws-cn:rds:" + testRegion + ":" + testAccountID + ":db:orders-db"},
+		{"unknown resource type", "arn:aws:rds:" + testRegion + ":" + testAccountID + ":cluster:orders"},
+		// Rejected at the ARN rather than at policy evaluation, so a foreign
+		// reference never reaches the evaluator at all.
+		{"foreign account", DBInstanceARN(testRegion, "210987654321", "orders-db")},
+		{"foreign region", DBInstanceARN("us-east-1", testAccountID, "orders-db")},
+		{"empty", ""},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			_, err := ParseARN(tc.arn, testRegion, testAccountID)
+			require.Error(t, err)
+			assert.Contains(t, err.Error(), awserrors.ErrorInvalidParameterValue)
+		})
+	}
+}
+
+// The registry is what makes an unregistered kind a rejection rather than an
+// empty tag list, so every kind the parser accepts has to be accounted for.
+func TestTaggableResources_OnlyDBInstancesAreRegistered(t *testing.T) {
+	_, ok := taggableResources[ResourceKindDBInstance]
+	assert.True(t, ok, "DB instances are the resource this phase tags")
+	for _, kind := range []ResourceKind{ResourceKindDBSnapshot, ResourceKindDBSubnetGroup, ResourceKindDBParameterGroup} {
+		_, ok := taggableResources[kind]
+		assert.False(t, ok, "%s is registered by the phase that creates it, not this one", kind)
+	}
+}

spinifex/handlers/rds/create.go

@@ -143,6 +143,7 @@ func newDBInstanceRecord(accountID string, req *validatedCreate, placement *endp
 		VpcID:                placement.VpcID,
 		VpcSecurityGroupIDs:  placement.SecurityGroupIDs,
 		DBParameterGroupName: req.DBParameterGroupName,
+		Tags:                 req.Tags,
 		Bootstrap:            BootstrapState{MasterUserPassword: req.MasterPassword},
 		CreatedAt:            now,
 		UpdatedAt:            now,

spinifex/handlers/rds/create_test.go

@@ -392,9 +392,6 @@ func TestValidateCreateRequest_RejectsUnimplementedParameters(t *testing.T) {
 		{"CloudwatchLogsExports", func(in *rds.CreateDBInstanceInput) {
 			in.EnableCloudwatchLogsExports = aws.StringSlice([]string{"postgresql"})
 		}, "EnableCloudwatchLogsExports"},
-		{"Tags", func(in *rds.CreateDBInstanceInput) {
-			in.Tags = []*rds.Tag{{Key: aws.String("env"), Value: aws.String("prod")}}
-		}, "Tags"},
 	}
 
 	for _, tc := range cases {

spinifex/handlers/rds/describe.go

@@ -86,6 +86,9 @@ func (s *Service) projectDBInstance(rec *DBInstanceRecord) *rds.DBInstance {
 		MultiAZ:              aws.Bool(false),
 		PubliclyAccessible:   aws.Bool(false),
 		InstanceCreateTime:   aws.Time(rec.CreatedAt),
+		// The Terraform provider reads tags from the describe as well as from
+		// ListTagsForResource, so the two have to agree.
+		TagList: tagsToAWS(rec.Tags),
 	}
 	if rec.DBName != "" {
 		out.DBName = aws.String(rec.DBName)

spinifex/handlers/rds/records.go

@@ -52,6 +52,10 @@ type DBInstanceRecord struct {
 	// stale reason cannot outlive the failure it describes.
 	FailureReason string `json:"failureReason,omitempty"`
 
+	// Inline rather than a separate key space, so the record delete that ends the
+	// instance also ends its tags.
+	Tags map[string]string `json:"tags,omitempty"`
+
 	Bootstrap BootstrapState `json:"bootstrap"`
 	Agent     AgentState     `json:"agent"`
 
@@ -59,6 +63,12 @@ type DBInstanceRecord struct {
 	UpdatedAt time.Time `json:"updatedAt"`
 }
 
+var _ TaggedRecord = (*DBInstanceRecord)(nil)
+
+func (r *DBInstanceRecord) GetTags() map[string]string { return r.Tags }
+
+func (r *DBInstanceRecord) SetTags(tags map[string]string) { r.Tags = tags }
+
 // The Consumed marker scopes the master password to a single fetch, not the
 // action: replace, recovery and restore all re-fetch without one.
 type BootstrapState struct {

spinifex/handlers/rds/service_nats.go

@@ -45,6 +45,21 @@ func (s *NATSService) DescribeDBInstances(ctx context.Context, input *rds.Descri
 		SubjectDescribeDBInstances, input, defaultTimeout, accountID)
 }
 
+func (s *NATSService) AddTagsToResource(ctx context.Context, input *rds.AddTagsToResourceInput, accountID string) (*rds.AddTagsToResourceOutput, error) {
+	return utils.NATSRequest[rds.AddTagsToResourceOutput](ctx, s.nc,
+		SubjectAddTagsToResource, input, defaultTimeout, accountID)
+}
+
+func (s *NATSService) RemoveTagsFromResource(ctx context.Context, input *rds.RemoveTagsFromResourceInput, accountID string) (*rds.RemoveTagsFromResourceOutput, error) {
+	return utils.NATSRequest[rds.RemoveTagsFromResourceOutput](ctx, s.nc,
+		SubjectRemoveTagsFromResource, input, defaultTimeout, accountID)
+}
+
+func (s *NATSService) ListTagsForResource(ctx context.Context, input *rds.ListTagsForResourceInput, accountID string) (*rds.ListTagsForResourceOutput, error) {
+	return utils.NATSRequest[rds.ListTagsForResourceOutput](ctx, s.nc,
+		SubjectListTagsForResource, input, defaultTimeout, accountID)
+}
+
 // Requested on the Layer-1 subject, not the bus.
 func (s *NATSService) GetDBBootstrapConfig(ctx context.Context, input *GetDBBootstrapConfigInput, accountID string) (*GetDBBootstrapConfigOutput, error) {
 	return utils.NATSRequest[GetDBBootstrapConfigOutput](ctx, s.nc,

spinifex/handlers/rds/subjects.go

@@ -22,6 +22,12 @@ const (
 	SubjectCreateDBInstance    = "rds.CreateDBInstance"
 	SubjectDescribeDBInstances = "rds.DescribeDBInstances"
 
+	// Tag actions address the resource by ARN, so one subject serves every
+	// resource type rather than one per kind.
+	SubjectAddTagsToResource      = "rds.AddTagsToResource"
+	SubjectRemoveTagsFromResource = "rds.RemoveTagsFromResource"
+	SubjectListTagsForResource    = "rds.ListTagsForResource"
+
 	// Queue groups are scoped per subject, so one name shares command delivery
 	// across gateway nodes for the whole fleet.
 	CommandQueueGroup = "spinifex-rds-agents"

spinifex/handlers/rds/tags.go

@@ -0,0 +1,259 @@
+package handlers_rds
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"maps"
+	"slices"
+	"strings"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/rds"
+	"github.com/mulgadc/spinifex/spinifex/awserrors"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// AWS's own tag limits. The reserved prefix is refused rather than dropped: a
+// silently discarded aws: tag would read back as absent on the next apply.
+const (
+	maxTagsPerResource = 50
+	maxTagKeyLen       = 128
+	maxTagValueLen     = 256
+	reservedTagPrefix  = "aws:"
+)
+
+// A tag write is a read-modify-write, so a concurrent add and remove race on
+// the same record. Each round has exactly one winner, so the bound is the
+// number of writers that can contend for one resource, not a timing guess.
+const tagWriteAttempts = 16
+
+// Implemented by every record whose ARN kind is registered below. Keeping the
+// tag map behind an interface is what lets one mutate path serve every
+// resource type without a type switch per action.
+type TaggedRecord interface {
+	GetTags() map[string]string
+	SetTags(tags map[string]string)
+}
+
+// How a resource kind's tags are reached: its key in the account bucket, the
+// record they live on, and the fault its absence raises.
+type taggableResource struct {
+	key       func(identifier string) string
+	newRecord func() TaggedRecord
+	notFound  string
+}
+
+// DB instances are the only taggable resource that exists so far. rds-7 adds
+// subnet groups and parameter groups here, rds-8 snapshots — one entry each,
+// registered by the phase that creates the resource.
+var taggableResources = map[ResourceKind]taggableResource{
+	ResourceKindDBInstance: {
+		key:       DBInstanceKey,
+		newRecord: func() TaggedRecord { return &DBInstanceRecord{} },
+		notFound:  awserrors.ErrorDBInstanceNotFound,
+	},
+}
+
+func (s *Service) ListTagsForResource(ctx context.Context, input *rds.ListTagsForResourceInput, accountID string) (*rds.ListTagsForResourceOutput, error) {
+	if input == nil {
+		return nil, fmt.Errorf("%s: empty request", awserrors.ErrorInvalidParameterValue)
+	}
+	resource, parsed, err := s.resolveTaggable(aws.StringValue(input.ResourceName), accountID)
+	if err != nil {
+		return nil, err
+	}
+	kv, err := s.bucket(ctx, accountID)
+	if err != nil {
+		return nil, err
+	}
+	rec, _, err := readTagged(ctx, kv, resource, parsed.Identifier)
+	if err != nil {
+		return nil, err
+	}
+	return &rds.ListTagsForResourceOutput{TagList: tagsToAWS(rec.GetTags())}, nil
+}
+
+// Adding a key that is already present overwrites it, so a repeated Terraform
+// apply converges instead of failing on the second run.
+func (s *Service) AddTagsToResource(ctx context.Context, input *rds.AddTagsToResourceInput, accountID string) (*rds.AddTagsToResourceOutput, error) {
+	if input == nil {
+		return nil, fmt.Errorf("%s: empty request", awserrors.ErrorInvalidParameterValue)
+	}
+	resource, parsed, err := s.resolveTaggable(aws.StringValue(input.ResourceName), accountID)
+	if err != nil {
+		return nil, err
+	}
+	add, err := validateTags(input.Tags)
+	if err != nil {
+		return nil, err
+	}
+	kv, err := s.bucket(ctx, accountID)
+	if err != nil {
+		return nil, err
+	}
+
+	err = mutateTags(ctx, kv, resource, parsed.Identifier, func(existing map[string]string) (map[string]string, error) {
+		merged := make(map[string]string, len(existing)+len(add))
+		maps.Copy(merged, existing)
+		maps.Copy(merged, add)
+		// Checked against the merge rather than the request: 40 existing tags plus
+		// 40 new ones is over the limit even though neither side is.
+		if len(merged) > maxTagsPerResource {
+			return nil, fmt.Errorf("%s: a resource may carry at most %d tags; this request would leave %d",
+				awserrors.ErrorInvalidParameterValue, maxTagsPerResource, len(merged))
+		}
+		return merged, nil
+	})
+	if err != nil {
+		return nil, err
+	}
+	return &rds.AddTagsToResourceOutput{}, nil
+}
+
+// Removing a key that is not present is a success, matching AWS: a retried
+// destroy must not fail on the tags it already removed.
+func (s *Service) RemoveTagsFromResource(ctx context.Context, input *rds.RemoveTagsFromResourceInput, accountID string) (*rds.RemoveTagsFromResourceOutput, error) {
+	if input == nil {
+		return nil, fmt.Errorf("%s: empty request", awserrors.ErrorInvalidParameterValue)
+	}
+	resource, parsed, err := s.resolveTaggable(aws.StringValue(input.ResourceName), accountID)
+	if err != nil {
+		return nil, err
+	}
+	keys := aws.StringValueSlice(input.TagKeys)
+	for _, key := range keys {
+		if err := validateTagKey(key); err != nil {
+			return nil, err
+		}
+	}
+	kv, err := s.bucket(ctx, accountID)
+	if err != nil {
+		return nil, err
+	}
+
+	err = mutateTags(ctx, kv, resource, parsed.Identifier, func(existing map[string]string) (map[string]string, error) {
+		remaining := maps.Clone(existing)
+		for _, key := range keys {
+			delete(remaining, key)
+		}
+		return remaining, nil
+	})
+	if err != nil {
+		return nil, err
+	}
+	return &rds.RemoveTagsFromResourceOutput{}, nil
+}
+
+// Parses the ARN and looks its kind up in the registry. A well-formed ARN for a
+// kind no phase has registered is rejected rather than answered with an empty
+// tag list, which would be indistinguishable from an untagged resource.
+func (s *Service) resolveTaggable(resourceName, accountID string) (taggableResource, ParsedARN, error) {
+	parsed, err := ParseARN(resourceName, s.region, accountID)
+	if err != nil {
+		return taggableResource{}, ParsedARN{}, err
+	}
+	resource, ok := taggableResources[parsed.Kind]
+	if !ok {
+		return taggableResource{}, ParsedARN{}, fmt.Errorf("%s: tagging %s resources is not supported yet",
+			awserrors.ErrorInvalidParameterValue, parsed.Kind)
+	}
+	return resource, parsed, nil
+}
+
+// The record plus its revision. A missing record raises the resource's own
+// not-found fault, so an ARN that parses but names nothing is distinguishable
+// from one that does not parse.
+func readTagged(ctx context.Context, kv jetstream.KeyValue, resource taggableResource, identifier string) (TaggedRecord, uint64, error) {
+	rec := resource.newRecord()
+	rev, found, err := getJSONRevision(ctx, kv, resource.key(identifier), rec)
+	if err != nil {
+		return nil, 0, err
+	}
+	if !found {
+		return nil, 0, errors.New(resource.notFound)
+	}
+	return rec, rev, nil
+}
+
+// Applies mutate to the record's tags under CAS. Losing the CAS means another
+// tag write landed first, so the mutation is replayed against its result rather
+// than overwriting it.
+func mutateTags(ctx context.Context, kv jetstream.KeyValue, resource taggableResource, identifier string, mutate func(map[string]string) (map[string]string, error)) error {
+	key := resource.key(identifier)
+	for range tagWriteAttempts {
+		rec, rev, err := readTagged(ctx, kv, resource, identifier)
+		if err != nil {
+			return err
+		}
+		next, err := mutate(rec.GetTags())
+		if err != nil {
+			return err
+		}
+		rec.SetTags(next)
+
+		err = updateJSON(ctx, kv, key, rev, rec)
+		if err == nil {
+			return nil
+		}
+		// jetstream.ErrKeyExists is a revision mismatch on Update, not a duplicate.
+		if !errors.Is(err, jetstream.ErrKeyExists) {
+			return err
+		}
+	}
+	return fmt.Errorf("rds: tag update on %s contended after %d attempts", key, tagWriteAttempts)
+}
+
+// The AWS tag rules, applied identically at create and at AddTagsToResource so
+// a create cannot produce tags a later modify would reject. A key repeated
+// within one request keeps its last value, as AWS does.
+func validateTags(tags []*rds.Tag) (map[string]string, error) {
+	if len(tags) > maxTagsPerResource {
+		return nil, fmt.Errorf("%s: at most %d tags may be supplied, got %d",
+			awserrors.ErrorInvalidParameterValue, maxTagsPerResource, len(tags))
+	}
+	out := make(map[string]string, len(tags))
+	for _, tag := range tags {
+		if tag == nil {
+			return nil, fmt.Errorf("%s: a tag entry is empty", awserrors.ErrorInvalidParameterValue)
+		}
+		key := aws.StringValue(tag.Key)
+		if err := validateTagKey(key); err != nil {
+			return nil, err
+		}
+		value := aws.StringValue(tag.Value)
+		if len(value) > maxTagValueLen {
+			return nil, fmt.Errorf("%s: tag value for key %q may be at most %d characters",
+				awserrors.ErrorInvalidParameterValue, key, maxTagValueLen)
+		}
+		out[key] = value
+	}
+	return out, nil
+}
+
+func validateTagKey(key string) error {
+	switch {
+	case key == "":
+		return fmt.Errorf("%s: a tag key may not be empty", awserrors.ErrorInvalidParameterValue)
+	case len(key) > maxTagKeyLen:
+		return fmt.Errorf("%s: tag key %q may be at most %d characters",
+			awserrors.ErrorInvalidParameterValue, key, maxTagKeyLen)
+	case strings.HasPrefix(strings.ToLower(key), reservedTagPrefix):
+		return fmt.Errorf("%s: tag key %q uses the reserved %q prefix",
+			awserrors.ErrorInvalidParameterValue, key, reservedTagPrefix)
+	}
+	return nil
+}
+
+// Sorted by key so a describe and a list return the same order on every call,
+// which is what keeps Terraform from reading a reshuffle as drift.
+func tagsToAWS(tags map[string]string) []*rds.Tag {
+	if len(tags) == 0 {
+		return nil
+	}
+	out := make([]*rds.Tag, 0, len(tags))
+	for _, key := range slices.Sorted(maps.Keys(tags)) {
+		out = append(out, &rds.Tag{Key: aws.String(key), Value: aws.String(tags[key])})
+	}
+	return out
+}

spinifex/handlers/rds/tags_test.go

@@ -0,0 +1,295 @@
+package handlers_rds
+
+import (
+	"fmt"
+	"strings"
+	"sync"
+	"testing"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/rds"
+	"github.com/mulgadc/spinifex/spinifex/awserrors"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func awsTags(kv ...string) []*rds.Tag {
+	tags := make([]*rds.Tag, 0, len(kv)/2)
+	for i := 0; i+1 < len(kv); i += 2 {
+		tags = append(tags, &rds.Tag{Key: aws.String(kv[i]), Value: aws.String(kv[i+1])})
+	}
+	return tags
+}
+
+// The tags on a DB instance as a map, read back through the public action so
+// the assertion covers the projection and not just the record.
+func listTags(t *testing.T, h *createHarness, id string) map[string]string {
+	t.Helper()
+	out, err := h.svc.ListTagsForResource(t.Context(), &rds.ListTagsForResourceInput{
+		ResourceName: aws.String(DBInstanceARN(testRegion, testAccountID, id)),
+	}, testAccountID)
+	require.NoError(t, err)
+	got := make(map[string]string, len(out.TagList))
+	for _, tag := range out.TagList {
+		got[aws.StringValue(tag.Key)] = aws.StringValue(tag.Value)
+	}
+	return got
+}
+
+func TestCreateDBInstance_TagsRoundTripThroughBothReadPaths(t *testing.T) {
+	h := newCreateHarness(t, "")
+	input := validCreateInput()
+	input.Tags = awsTags("env", "prod", "team", "platform")
+
+	out, err := h.svc.CreateDBInstance(t.Context(), input, testAccountID)
+	require.NoError(t, err)
+
+	// Terraform reads tags from both the describe and the list, so the two
+	// disagreeing would show up as permanent drift.
+	assert.Equal(t, input.Tags, out.DBInstance.TagList)
+	described, err := h.svc.DescribeDBInstances(t.Context(), &rds.DescribeDBInstancesInput{
+		DBInstanceIdentifier: aws.String(testDBInstanceID),
+	}, testAccountID)
+	require.NoError(t, err)
+	require.Len(t, described.DBInstances, 1)
+	assert.Equal(t, input.Tags, described.DBInstances[0].TagList)
+
+	assert.Equal(t, map[string]string{"env": "prod", "team": "platform"}, listTags(t, h, testDBInstanceID))
+}
+
+// The same rules apply at create as at AddTagsToResource, and they run before
+// the identifier is reserved so a rejected create leaves nothing behind.
+func TestCreateDBInstance_InvalidTagsLeaveNoRecord(t *testing.T) {
+	h := newCreateHarness(t, "")
+	input := validCreateInput()
+	input.Tags = awsTags("aws:cloudformation:stack-name", "mine")
+
+	_, err := h.svc.CreateDBInstance(t.Context(), input, testAccountID)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), awserrors.ErrorInvalidParameterValue)
+	assert.False(t, h.recordExists(t, testDBInstanceID),
+		"a create rejected on its tags must not reserve the identifier")
+}
+
+func TestListTagsForResource_UntaggedInstanceIsAnEmptyList(t *testing.T) {
+	h := newCreateHarness(t, "")
+	seedCreated(t, h, testDBInstanceID)
+
+	assert.Empty(t, listTags(t, h, testDBInstanceID))
+}
+
+// AddTagsToResource is a merge, and a repeated key overwrites rather than
+// duplicates, so a re-run of the same apply converges.
+func TestAddTagsToResource_MergesAndOverwrites(t *testing.T) {
+	h := newCreateHarness(t, "")
+	seedCreated(t, h, testDBInstanceID)
+	arn := aws.String(DBInstanceARN(testRegion, testAccountID, testDBInstanceID))
+
+	_, err := h.svc.AddTagsToResource(t.Context(), &rds.AddTagsToResourceInput{
+		ResourceName: arn, Tags: awsTags("env", "staging", "team", "platform"),
+	}, testAccountID)
+	require.NoError(t, err)
+
+	_, err = h.svc.AddTagsToResource(t.Context(), &rds.AddTagsToResourceInput{
+		ResourceName: arn, Tags: awsTags("env", "prod"),
+	}, testAccountID)
+	require.NoError(t, err)
+
+	assert.Equal(t, map[string]string{"env": "prod", "team": "platform"}, listTags(t, h, testDBInstanceID))
+}
+
+// A key repeated inside one request keeps its last value rather than failing,
+// which is what AWS does.
+func TestAddTagsToResource_DuplicateKeyInOneRequestKeepsTheLastValue(t *testing.T) {
+	h := newCreateHarness(t, "")
+	seedCreated(t, h, testDBInstanceID)
+
+	_, err := h.svc.AddTagsToResource(t.Context(), &rds.AddTagsToResourceInput{
+		ResourceName: aws.String(DBInstanceARN(testRegion, testAccountID, testDBInstanceID)),
+		Tags:         awsTags("env", "staging", "env", "prod"),
+	}, testAccountID)
+	require.NoError(t, err)
+
+	assert.Equal(t, map[string]string{"env": "prod"}, listTags(t, h, testDBInstanceID))
+}
+
+func TestRemoveTagsFromResource_RemovesNamedKeysAndIgnoresAbsentOnes(t *testing.T) {
+	h := newCreateHarness(t, "")
+	seedCreated(t, h, testDBInstanceID)
+	arn := aws.String(DBInstanceARN(testRegion, testAccountID, testDBInstanceID))
+
+	_, err := h.svc.AddTagsToResource(t.Context(), &rds.AddTagsToResourceInput{
+		ResourceName: arn, Tags: awsTags("env", "prod", "team", "platform"),
+	}, testAccountID)
+	require.NoError(t, err)
+
+	// "owner" was never set: a destroy that already ran must not fail on its
+	// second attempt.
+	_, err = h.svc.RemoveTagsFromResource(t.Context(), &rds.RemoveTagsFromResourceInput{
+		ResourceName: arn, TagKeys: aws.StringSlice([]string{"env", "owner"}),
+	}, testAccountID)
+	require.NoError(t, err)
+
+	assert.Equal(t, map[string]string{"team": "platform"}, listTags(t, h, testDBInstanceID))
+}
+
+func TestTagActions_RejectInvalidTags(t *testing.T) {
+	overLimit := make([]string, 0, (maxTagsPerResource+1)*2)
+	for i := 0; i <= maxTagsPerResource; i++ {
+		overLimit = append(overLimit, fmt.Sprintf("key-%d", i), "v")
+	}
+
+	cases := []struct {
+		name string
+		tags []*rds.Tag
+	}{
+		{"over the tag limit", awsTags(overLimit...)},
+		{"oversized key", awsTags(strings.Repeat("k", maxTagKeyLen+1), "v")},
+		{"oversized value", awsTags("env", strings.Repeat("v", maxTagValueLen+1))},
+		{"reserved prefix", awsTags("aws:createdBy", "me")},
+		{"reserved prefix in any case", awsTags("AWS:createdBy", "me")},
+		{"empty key", awsTags("", "v")},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			h := newCreateHarness(t, "")
+			seedCreated(t, h, testDBInstanceID)
+
+			_, err := h.svc.AddTagsToResource(t.Context(), &rds.AddTagsToResourceInput{
+				ResourceName: aws.String(DBInstanceARN(testRegion, testAccountID, testDBInstanceID)),
+				Tags:         tc.tags,
+			}, testAccountID)
+			require.Error(t, err)
+			assert.Contains(t, err.Error(), awserrors.ErrorInvalidParameterValue)
+			assert.Empty(t, listTags(t, h, testDBInstanceID), "a rejected request must write nothing")
+		})
+	}
+}
+
+// The limit is a per-resource one, so two individually legal requests that
+// together exceed it are rejected on the second.
+func TestAddTagsToResource_LimitAppliesToTheMergedResult(t *testing.T) {
+	h := newCreateHarness(t, "")
+	seedCreated(t, h, testDBInstanceID)
+	arn := aws.String(DBInstanceARN(testRegion, testAccountID, testDBInstanceID))
+
+	half := make([]string, 0, maxTagsPerResource)
+	for i := range maxTagsPerResource / 2 {
+		half = append(half, fmt.Sprintf("first-%d", i), "v")
+	}
+	_, err := h.svc.AddTagsToResource(t.Context(), &rds.AddTagsToResourceInput{
+		ResourceName: arn, Tags: awsTags(half...),
+	}, testAccountID)
+	require.NoError(t, err)
+
+	second := make([]string, 0, maxTagsPerResource)
+	for i := 0; i <= maxTagsPerResource/2; i++ {
+		second = append(second, fmt.Sprintf("second-%d", i), "v")
+	}
+	_, err = h.svc.AddTagsToResource(t.Context(), &rds.AddTagsToResourceInput{
+		ResourceName: arn, Tags: awsTags(second...),
+	}, testAccountID)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), awserrors.ErrorInvalidParameterValue)
+	assert.Len(t, listTags(t, h, testDBInstanceID), maxTagsPerResource/2,
+		"the rejected merge must leave the existing tags untouched")
+}
+
+// A well-formed ARN naming nothing is the resource's own fault, not an ARN
+// error: the two failures carry different messages on purpose.
+func TestTagActions_MissingResourceIsTheResourcesOwnFault(t *testing.T) {
+	h := newCreateHarness(t, "")
+	arn := aws.String(DBInstanceARN(testRegion, testAccountID, "no-such-db"))
+
+	_, err := h.svc.ListTagsForResource(t.Context(), &rds.ListTagsForResourceInput{ResourceName: arn}, testAccountID)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), awserrors.ErrorDBInstanceNotFound)
+
+	_, err = h.svc.AddTagsToResource(t.Context(), &rds.AddTagsToResourceInput{
+		ResourceName: arn, Tags: awsTags("env", "prod"),
+	}, testAccountID)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), awserrors.ErrorDBInstanceNotFound)
+
+	_, err = h.svc.RemoveTagsFromResource(t.Context(), &rds.RemoveTagsFromResourceInput{
+		ResourceName: arn, TagKeys: aws.StringSlice([]string{"env"}),
+	}, testAccountID)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), awserrors.ErrorDBInstanceNotFound)
+}
+
+// Snapshots are a valid ARN kind that no phase has built yet. Answering with an
+// empty list would be indistinguishable from an untagged snapshot and would let
+// an apply appear to tag something that does not exist.
+func TestTagActions_UnregisteredResourceTypeIsRejected(t *testing.T) {
+	h := newCreateHarness(t, "")
+
+	_, err := h.svc.ListTagsForResource(t.Context(), &rds.ListTagsForResourceInput{
+		ResourceName: aws.String(DBSnapshotARN(testRegion, testAccountID, "orders-db-snap")),
+	}, testAccountID)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), awserrors.ErrorInvalidParameterValue)
+}
+
+func TestTagActions_ForeignAccountARNIsRejected(t *testing.T) {
+	h := newCreateHarness(t, "")
+	seedCreated(t, h, testDBInstanceID)
+
+	_, err := h.svc.ListTagsForResource(t.Context(), &rds.ListTagsForResourceInput{
+		ResourceName: aws.String(DBInstanceARN(testRegion, "210987654321", testDBInstanceID)),
+	}, testAccountID)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), awserrors.ErrorInvalidParameterValue)
+}
+
+// Every tag write is a read-modify-write of the same record, so without CAS the
+// last writer would silently discard the others' keys.
+func TestTagWrites_ConcurrentAddsAndRemovesDoNotLoseEachOther(t *testing.T) {
+	h := newCreateHarness(t, "")
+	seedCreated(t, h, testDBInstanceID)
+	arn := aws.String(DBInstanceARN(testRegion, testAccountID, testDBInstanceID))
+
+	// Seeded first so the concurrent removes have something to take away.
+	const writers = 4
+	seed := make([]string, 0, writers*2)
+	for i := range writers {
+		seed = append(seed, fmt.Sprintf("doomed-%d", i), "v")
+	}
+	_, err := h.svc.AddTagsToResource(t.Context(), &rds.AddTagsToResourceInput{
+		ResourceName: arn, Tags: awsTags(seed...),
+	}, testAccountID)
+	require.NoError(t, err)
+
+	var wg sync.WaitGroup
+	errs := make(chan error, writers*2)
+	for i := range writers {
+		wg.Add(2)
+		go func() {
+			defer wg.Done()
+			_, err := h.svc.AddTagsToResource(t.Context(), &rds.AddTagsToResourceInput{
+				ResourceName: arn, Tags: awsTags(fmt.Sprintf("kept-%d", i), "v"),
+			}, testAccountID)
+			errs <- err
+		}()
+		go func() {
+			defer wg.Done()
+			_, err := h.svc.RemoveTagsFromResource(t.Context(), &rds.RemoveTagsFromResourceInput{
+				ResourceName: arn, TagKeys: aws.StringSlice([]string{fmt.Sprintf("doomed-%d", i)}),
+			}, testAccountID)
+			errs <- err
+		}()
+	}
+	wg.Wait()
+	close(errs)
+	for err := range errs {
+		require.NoError(t, err)
+	}
+
+	got := listTags(t, h, testDBInstanceID)
+	want := map[string]string{}
+	for i := range writers {
+		want[fmt.Sprintf("kept-%d", i)] = "v"
+	}
+	assert.Equal(t, want, got, "every add must survive and every remove must take effect")
+}

spinifex/handlers/rds/validate.go

@@ -48,6 +48,7 @@ type validatedCreate struct {
 	// this with a real DB subnet group lookup.
 	SubnetID             string
 	DBParameterGroupName string
+	Tags                 map[string]string
 }
 
 // Everything that can be decided from the request alone. Network resolution
@@ -128,6 +129,13 @@ func validateCreateRequest(input *rds.CreateDBInstanceInput) (*validatedCreate,
 		return nil, fmt.Errorf("%s: DB parameter group %q not found", awserrors.ErrorDBParameterGroupNotFound, paramGroup)
 	}
 
+	// Rejected before the identifier is reserved, so a create with bad tags
+	// leaves no partial record behind.
+	tags, err := validateTags(input.Tags)
+	if err != nil {
+		return nil, err
+	}
+
 	return &validatedCreate{
 		Identifier:           identifier,
 		Engine:               engine,
@@ -142,6 +150,7 @@ func validateCreateRequest(input *rds.CreateDBInstanceInput) (*validatedCreate,
 		DBName:               aws.StringValue(input.DBName),
 		SecurityGroupIDs:     aws.StringValueSlice(input.VpcSecurityGroupIds),
 		DBParameterGroupName: engine.DefaultParameterGroupName(),
+		Tags:                 tags,
 	}, nil
 }
 
@@ -222,9 +231,6 @@ func rejectUnimplemented(input *rds.CreateDBInstanceInput) error {
 	if len(input.EnableCloudwatchLogsExports) > 0 {
 		return unimplemented("EnableCloudwatchLogsExports", "log export is not implemented")
 	}
-	if len(input.Tags) > 0 {
-		return unimplemented("Tags", "resource tagging is not implemented yet")
-	}
 	return nil
 }
 

tests/e2e/rds/create_describe_test.go

@@ -48,6 +48,10 @@ func TestCreateDescribeConnect(t *testing.T) {
 		DBName:               aws.String(dbName),
 		MasterUsername:       aws.String(dbMasterUser),
 		MasterUserPassword:   aws.String(dbMasterPassword),
+		Tags: []*rds.Tag{
+			{Key: aws.String("env"), Value: aws.String("e2e")},
+			{Key: aws.String("suite"), Value: aws.String("rds")},
+		},
 	})
 	require.NoError(t, err, "create-db-instance")
 	require.NotNil(t, out.DBInstance)
@@ -87,6 +91,44 @@ func TestCreateDescribeConnect(t *testing.T) {
 		assert.Contains(t, ids, id, "an unfiltered describe must list the instance")
 	})
 
+	// The Terraform provider reads tags through ListTagsForResource on every
+	// apply, so this is the assertion that proves the provider's read path works.
+	t.Run("TagsRoundTrip", func(t *testing.T) {
+		requireAvailable(t, instance)
+		arn := aws.StringValue(instance.DBInstanceArn)
+		require.NotEmpty(t, arn, "an available instance must publish its ARN")
+
+		tags, err := f.AWS.RDS.ListTagsForResource(&rds.ListTagsForResourceInput{ResourceName: aws.String(arn)})
+		require.NoError(t, err, "list-tags-for-resource")
+		assert.Equal(t, map[string]string{"env": "e2e", "suite": "rds"}, tagMap(tags.TagList),
+			"the tags supplied at create must be readable back")
+
+		_, err = f.AWS.RDS.AddTagsToResource(&rds.AddTagsToResourceInput{
+			ResourceName: aws.String(arn),
+			Tags:         []*rds.Tag{{Key: aws.String("env"), Value: aws.String("e2e-updated")}},
+		})
+		require.NoError(t, err, "add-tags-to-resource")
+
+		_, err = f.AWS.RDS.RemoveTagsFromResource(&rds.RemoveTagsFromResourceInput{
+			ResourceName: aws.String(arn),
+			TagKeys:      aws.StringSlice([]string{"suite", "never-set"}),
+		})
+		require.NoError(t, err, "remove-tags-from-resource must ignore a key that is not present")
+
+		tags, err = f.AWS.RDS.ListTagsForResource(&rds.ListTagsForResourceInput{ResourceName: aws.String(arn)})
+		require.NoError(t, err, "list-tags-for-resource")
+		assert.Equal(t, map[string]string{"env": "e2e-updated"}, tagMap(tags.TagList))
+
+		// Terraform reads tags from the describe as well, so the two views
+		// disagreeing would show up as permanent drift.
+		described, err := f.AWS.RDS.DescribeDBInstances(&rds.DescribeDBInstancesInput{
+			DBInstanceIdentifier: aws.String(id),
+		})
+		require.NoError(t, err, "describe-db-instances")
+		require.Len(t, described.DBInstances, 1)
+		assert.Equal(t, map[string]string{"env": "e2e-updated"}, tagMap(described.DBInstances[0].TagList))
+	})
+
 	t.Run("EndpointResolves", func(t *testing.T) {
 		requireAvailable(t, instance)
 		host := aws.StringValue(instance.Endpoint.Address)
@@ -109,6 +151,14 @@ func TestCreateDescribeConnect(t *testing.T) {
 	})
 }
 
+func tagMap(tags []*rds.Tag) map[string]string {
+	out := make(map[string]string, len(tags))
+	for _, tag := range tags {
+		out[aws.StringValue(tag.Key)] = aws.StringValue(tag.Value)
+	}
+	return out
+}
+
 func requireAvailable(t *testing.T, instance *rds.DBInstance) {
 	t.Helper()
 	if instance == nil {