fix(ec2): clear a resource's central tags when it is deleted (#1073)

DescribeTags kept answering for resources that no longer exist: the central tag store's DeleteAllTags had one production caller, the instance terminate path, so every other resource type left its entry behind forever. CentralTagWriter becomes CentralTagStore and gains DeleteAllTags, which TagsServiceImpl already implemented, so the daemon wiring is a rename rather than new plumbing. clearRecordTags mirrors projectRecordTags: a nil store is a no-op and a failure is logged rather than returned. The clear runs after the record delete has succeeded, never before, so a delete that fails leaves a live resource with its tags intact. deleteNetworkInterface covers both the public delete and the forced instance teardown, and deleteSecurityGroupInternal covers the default group that only the DeleteVpc cascade ever removes.
2026-10-15 17:11:30 +00:00 · 2026-09-17 15:18:31 +10:00
parent e88691e64d
commit c9460628fb
7 changed files with 245 additions and 38 deletions
@@ -1651,9 +1651,10 @@ func (d *Daemon) startCluster() error {
 	// pools that hand out routable public IPs (pool mode always; nat mode only
 	// when a public pool rides alongside the transit segment).
 	d.vpcService.SetDefaultPublicIPMapping(d.hasPublicIPPools())
-	// Project vpc/subnet/sg/eni creation-time tags into the central tag store
-	// so describe-tags agrees with each resource's own describe from birth.
-	d.vpcService.SetCentralTagWriter(d.tagsService)
+	// Keep the central tag store in step with vpc/subnet/sg/eni records, so
+	// describe-tags agrees with each resource's own describe from birth and
+	// stops answering for it once it is deleted.
+	d.vpcService.SetCentralTagStore(d.tagsService)

 	d.routeTableService, err = initServiceWithRetry("RouteTable service", func() (*handlers_ec2_routetable.RouteTableServiceImpl, error) {
 		return handlers_ec2_routetable.NewRouteTableServiceImplWithNATS(d.ctx, d.config, d.natsConn)
@@ -32,8 +32,8 @@ var _ TagsService = (*TagsServiceImpl)(nil)
 // Ensure TagsServiceImpl can project instance record tags into the store.
 var _ handlers_ec2_instance.InstanceTagWriter = (*TagsServiceImpl)(nil)

-// Ensure TagsServiceImpl can project vpc/subnet/sg/eni record tags into the store.
-var _ handlers_ec2_vpc.CentralTagWriter = (*TagsServiceImpl)(nil)
+// Ensure TagsServiceImpl can both project and clear vpc/subnet/sg/eni record tags.
+var _ handlers_ec2_vpc.CentralTagStore = (*TagsServiceImpl)(nil)

 // TagsServiceImpl implements TagsService over a JetStream KV bucket, one entry
 // per resource, scoped by account in the key.
@@ -183,9 +183,10 @@ func (s *TagsServiceImpl) PutResourceTags(ctx context.Context, accountID, resour
 	return s.putTags(ctx, accountID, resourceID, tags)
 }

-// DeleteAllTags removes the stored tags for a resource. Used on instance
-// terminate so describe-tags stops reporting the instance while the terminated
-// record keeps its tags until TTL. Idempotent: an absent entry is not an error.
+// DeleteAllTags removes the stored tags for a resource, so describe-tags stops
+// reporting a resource that is gone. Instance terminate uses it while the
+// terminated record keeps its own tags until TTL; the vpc delete paths use it
+// to retire the entry outright. Idempotent: an absent entry is not an error.
 func (s *TagsServiceImpl) DeleteAllTags(ctx context.Context, accountID, resourceID string) error {
 	if err := s.deleteTagsEntry(ctx, accountID, resourceID); err != nil {
 		return err
@@ -11,15 +11,16 @@ import (
 	"github.com/stretchr/testify/require"
 )

-// fakeCentralTagWriter records the tag map passed to PutResourceTags per
-// resource, and can be made to fail so tests can prove a central-store error
-// never fails the create it is projecting from.
-type fakeCentralTagWriter struct {
-	calls map[string]map[string]string
-	err   error
+// fakeCentralTagStore records the tag map passed to PutResourceTags per
+// resource and the ids passed to DeleteAllTags, and can be made to fail so
+// tests can prove a central-store error never fails the operation it tracks.
+type fakeCentralTagStore struct {
+	calls   map[string]map[string]string
+	deleted []string
+	err     error
 }

-func (f *fakeCentralTagWriter) PutResourceTags(_ context.Context, _, resourceID string, tags map[string]string) error {
+func (f *fakeCentralTagStore) PutResourceTags(_ context.Context, _, resourceID string, tags map[string]string) error {
 	if f.calls == nil {
 		f.calls = map[string]map[string]string{}
 	}
@@ -27,11 +28,16 @@ func (f *fakeCentralTagWriter) PutResourceTags(_ context.Context, _, resourceID
 	return f.err
 }

+func (f *fakeCentralTagStore) DeleteAllTags(_ context.Context, _, resourceID string) error {
+	f.deleted = append(f.deleted, resourceID)
+	return f.err
+}
+
 func TestCreateVpc_ProjectsTagsToCentralStore(t *testing.T) {
 	t.Parallel()
 	svc := setupTestVPCService(t)
-	writer := &fakeCentralTagWriter{}
-	svc.SetCentralTagWriter(writer)
+	writer := &fakeCentralTagStore{}
+	svc.SetCentralTagStore(writer)

 	out, err := svc.CreateVpc(context.Background(), &ec2.CreateVpcInput{
 		CidrBlock: aws.String("10.0.0.0/16"),
@@ -54,8 +60,8 @@ func TestCreateVpc_ProjectsTagsToCentralStore(t *testing.T) {
 func TestCreateSubnet_ProjectsTagsToCentralStore(t *testing.T) {
 	t.Parallel()
 	svc := setupTestVPCService(t)
-	writer := &fakeCentralTagWriter{}
-	svc.SetCentralTagWriter(writer)
+	writer := &fakeCentralTagStore{}
+	svc.SetCentralTagStore(writer)
 	vpcID := createTestVPC(t, svc, "10.0.0.0/16")

 	out, err := svc.CreateSubnet(context.Background(), &ec2.CreateSubnetInput{
@@ -80,8 +86,8 @@ func TestCreateSubnet_ProjectsTagsToCentralStore(t *testing.T) {
 func TestCreateSecurityGroup_ProjectsTagsToCentralStore(t *testing.T) {
 	t.Parallel()
 	svc := setupTestVPCService(t)
-	writer := &fakeCentralTagWriter{}
-	svc.SetCentralTagWriter(writer)
+	writer := &fakeCentralTagStore{}
+	svc.SetCentralTagStore(writer)
 	vpcID := createTestVPC(t, svc, "10.0.0.0/16")

 	out, err := svc.CreateSecurityGroup(context.Background(), &ec2.CreateSecurityGroupInput{
@@ -107,8 +113,8 @@ func TestCreateSecurityGroup_ProjectsTagsToCentralStore(t *testing.T) {
 func TestCreateVpc_NoTags_SkipsCentralStoreWrite(t *testing.T) {
 	t.Parallel()
 	svc := setupTestVPCService(t)
-	writer := &fakeCentralTagWriter{}
-	svc.SetCentralTagWriter(writer)
+	writer := &fakeCentralTagStore{}
+	svc.SetCentralTagStore(writer)

 	out, err := svc.CreateVpc(context.Background(), &ec2.CreateVpcInput{
 		CidrBlock: aws.String("10.0.0.0/16"),
@@ -118,10 +124,10 @@ func TestCreateVpc_NoTags_SkipsCentralStoreWrite(t *testing.T) {
 	assert.NotContains(t, writer.calls, *out.Vpc.VpcId)
 }

-func TestCreateVpc_NilCentralTagWriter_CreateSucceeds(t *testing.T) {
+func TestCreateVpc_NilCentralTagStore_CreateSucceeds(t *testing.T) {
 	t.Parallel()
 	svc := setupTestVPCService(t)
-	// No SetCentralTagWriter call: centralTags stays nil.
+	// No SetCentralTagStore call: centralTags stays nil.

 	out, err := svc.CreateVpc(context.Background(), &ec2.CreateVpcInput{
 		CidrBlock: aws.String("10.0.0.0/16"),
@@ -138,11 +144,11 @@ func TestCreateVpc_NilCentralTagWriter_CreateSucceeds(t *testing.T) {
 	require.Len(t, out.Vpc.Tags, 1)
 }

-func TestCreateVpc_CentralTagWriterError_DoesNotFailCreate(t *testing.T) {
+func TestCreateVpc_CentralTagStoreError_DoesNotFailCreate(t *testing.T) {
 	t.Parallel()
 	svc := setupTestVPCService(t)
-	writer := &fakeCentralTagWriter{err: errors.New("central store unavailable")}
-	svc.SetCentralTagWriter(writer)
+	writer := &fakeCentralTagStore{err: errors.New("central store unavailable")}
+	svc.SetCentralTagStore(writer)

 	out, err := svc.CreateVpc(context.Background(), &ec2.CreateVpcInput{
 		CidrBlock: aws.String("10.0.0.0/16"),
@@ -158,3 +164,168 @@ func TestCreateVpc_CentralTagWriterError_DoesNotFailCreate(t *testing.T) {
 	require.NoError(t, err)
 	require.Len(t, out.Vpc.Tags, 1)
 }
+
+// Deleting a resource must retire its central tag entry, whatever the type.
+// The table drives each delete path through the same store so the assertion is
+// about the shared teardown rather than one handler: a type added to the
+// package without a clear call fails here rather than leaking quietly.
+func TestDelete_ClearsCentralTagStore(t *testing.T) {
+	t.Parallel()
+
+	cases := []struct {
+		name string
+		// create returns the id of a resource of this type, already tagged.
+		create func(t *testing.T, svc *VPCServiceImpl) string
+		del    func(t *testing.T, svc *VPCServiceImpl, id string)
+	}{
+		{
+			name: "security group",
+			create: func(t *testing.T, svc *VPCServiceImpl) string {
+				vpcID := createTestVPC(t, svc, "10.1.0.0/16")
+				out, err := svc.CreateSecurityGroup(context.Background(), &ec2.CreateSecurityGroupInput{
+					GroupName:   aws.String("doomed-sg"),
+					Description: aws.String("doomed"),
+					VpcId:       aws.String(vpcID),
+					TagSpecifications: []*ec2.TagSpecification{
+						{ResourceType: aws.String("security-group"), Tags: []*ec2.Tag{
+							{Key: aws.String("Name"), Value: aws.String("doomed")},
+						}},
+					},
+				}, testAccountID)
+				require.NoError(t, err)
+				return *out.GroupId
+			},
+			del: func(t *testing.T, svc *VPCServiceImpl, id string) {
+				_, err := svc.DeleteSecurityGroup(context.Background(),
+					&ec2.DeleteSecurityGroupInput{GroupId: aws.String(id)}, testAccountID)
+				require.NoError(t, err)
+			},
+		},
+		{
+			name: "subnet",
+			create: func(t *testing.T, svc *VPCServiceImpl) string {
+				vpcID := createTestVPC(t, svc, "10.2.0.0/16")
+				out, err := svc.CreateSubnet(context.Background(), &ec2.CreateSubnetInput{
+					VpcId:     aws.String(vpcID),
+					CidrBlock: aws.String("10.2.1.0/24"),
+					TagSpecifications: []*ec2.TagSpecification{
+						{ResourceType: aws.String("subnet"), Tags: []*ec2.Tag{
+							{Key: aws.String("Name"), Value: aws.String("doomed")},
+						}},
+					},
+				}, testAccountID)
+				require.NoError(t, err)
+				return *out.Subnet.SubnetId
+			},
+			del: func(t *testing.T, svc *VPCServiceImpl, id string) {
+				_, err := svc.DeleteSubnet(context.Background(),
+					&ec2.DeleteSubnetInput{SubnetId: aws.String(id)}, testAccountID)
+				require.NoError(t, err)
+			},
+		},
+		{
+			// The interface that first exposed this gap. Both the public delete
+			// and the forced instance teardown run the same path, so covering
+			// one covers the ENI reclaimed by a task going away.
+			name: "network interface",
+			create: func(t *testing.T, svc *VPCServiceImpl) string {
+				vpcID := createTestVPC(t, svc, "10.7.0.0/16")
+				subnetID := createTestSubnet(t, svc, vpcID, "10.7.1.0/24")
+				out, err := svc.CreateNetworkInterface(context.Background(), &ec2.CreateNetworkInterfaceInput{
+					SubnetId: aws.String(subnetID),
+					TagSpecifications: []*ec2.TagSpecification{
+						{ResourceType: aws.String("network-interface"), Tags: []*ec2.Tag{
+							{Key: aws.String("Name"), Value: aws.String("doomed")},
+						}},
+					},
+				}, testAccountID)
+				require.NoError(t, err)
+				return *out.NetworkInterface.NetworkInterfaceId
+			},
+			del: func(t *testing.T, svc *VPCServiceImpl, id string) {
+				_, err := svc.DeleteNetworkInterface(context.Background(),
+					&ec2.DeleteNetworkInterfaceInput{NetworkInterfaceId: aws.String(id)}, testAccountID)
+				require.NoError(t, err)
+			},
+		},
+		{
+			name: "vpc",
+			create: func(t *testing.T, svc *VPCServiceImpl) string {
+				out, err := svc.CreateVpc(context.Background(), &ec2.CreateVpcInput{
+					CidrBlock: aws.String("10.3.0.0/16"),
+					TagSpecifications: []*ec2.TagSpecification{
+						{ResourceType: aws.String("vpc"), Tags: []*ec2.Tag{
+							{Key: aws.String("Name"), Value: aws.String("doomed")},
+						}},
+					},
+				}, testAccountID)
+				require.NoError(t, err)
+				return *out.Vpc.VpcId
+			},
+			del: func(t *testing.T, svc *VPCServiceImpl, id string) {
+				_, err := svc.DeleteVpc(context.Background(),
+					&ec2.DeleteVpcInput{VpcId: aws.String(id)}, testAccountID)
+				require.NoError(t, err)
+			},
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			svc := setupTestVPCService(t)
+			store := &fakeCentralTagStore{}
+			svc.SetCentralTagStore(store)
+
+			id := tc.create(t, svc)
+			require.Contains(t, store.calls, id, "precondition: the create must have tagged it")
+
+			tc.del(t, svc, id)
+			assert.Contains(t, store.deleted, id, "the deleted resource's tags must be retired")
+		})
+	}
+}
+
+// DeleteVpc cascades the default security group, which no caller ever deletes
+// by hand. Asserted separately because the id is not the one under test.
+func TestDeleteVpc_ClearsTheCascadedDefaultSecurityGroup(t *testing.T) {
+	t.Parallel()
+	svc := setupTestVPCService(t)
+	store := &fakeCentralTagStore{}
+	svc.SetCentralTagStore(store)
+
+	vpcID := createTestVPC(t, svc, "10.4.0.0/16")
+	defaultSG, err := svc.FindDefaultSGForVPC(testAccountID, vpcID)
+	require.NoError(t, err)
+	require.NotEmpty(t, defaultSG, "precondition: CreateVpc makes a default SG")
+
+	_, err = svc.DeleteVpc(context.Background(), &ec2.DeleteVpcInput{VpcId: aws.String(vpcID)}, testAccountID)
+	require.NoError(t, err)
+
+	assert.Contains(t, store.deleted, defaultSG, "the cascaded default SG must not keep its tags")
+}
+
+func TestDelete_NilCentralTagStore_DeleteSucceeds(t *testing.T) {
+	t.Parallel()
+	svc := setupTestVPCService(t)
+	// No SetCentralTagStore call: centralTags stays nil.
+
+	vpcID := createTestVPC(t, svc, "10.5.0.0/16")
+	_, err := svc.DeleteVpc(context.Background(), &ec2.DeleteVpcInput{VpcId: aws.String(vpcID)}, testAccountID)
+	require.NoError(t, err)
+}
+
+// The resource is already gone by the time the store is called, so a store
+// failure must not fail the delete: the caller cannot retry something that has
+// happened, and would be left believing the resource survived.
+func TestDelete_CentralTagStoreError_DoesNotFailDelete(t *testing.T) {
+	t.Parallel()
+	svc := setupTestVPCService(t)
+	store := &fakeCentralTagStore{err: errors.New("central store unavailable")}
+	svc.SetCentralTagStore(store)
+
+	vpcID := createTestVPC(t, svc, "10.6.0.0/16")
+	_, err := svc.DeleteVpc(context.Background(), &ec2.DeleteVpcInput{VpcId: aws.String(vpcID)}, testAccountID)
+	require.NoError(t, err)
+	assert.Contains(t, store.deleted, vpcID, "the clear must still have been attempted")
+}
@@ -275,6 +275,10 @@ func (s *VPCServiceImpl) deleteNetworkInterface(ctx context.Context, eniId, acco
 	if err := s.eniKV.Delete(ctx, key); err != nil {
 		return nil, errors.New(awserrors.ErrorServerInternal)
 	}
+	// Both the public delete and the forced instance teardown land here, which
+	// is what covers an ENI reclaimed by a task or instance going away rather
+	// than by an API call.
+	s.clearRecordTags(ctx, accountID, eniId)

 	slog.InfoContext(ctx, "DeleteNetworkInterface completed", "eniId", eniId, "accountID", accountID)

@@ -302,6 +302,7 @@ func (s *VPCServiceImpl) DeleteSecurityGroup(ctx context.Context, input *ec2.Del
 	if err := s.sgKV.Delete(ctx, key); err != nil {
 		return nil, errors.New(awserrors.ErrorServerInternal)
 	}
+	s.clearRecordTags(ctx, accountID, groupId)

 	slog.InfoContext(ctx, "DeleteSecurityGroup completed", "groupId", groupId, "accountID", accountID)

@@ -1714,6 +1715,11 @@ func (s *VPCServiceImpl) deleteSecurityGroupInternal(ctx context.Context, accoun
 	if err := s.sgKV.Delete(ctx, key); err != nil {
 		return fmt.Errorf("delete default security group: %w", err)
 	}
+	// The cascade from DeleteVpc reaches the default SG only through here, so
+	// without this the one group the caller never deletes by hand is the one
+	// that always leaks its tags.
+	s.clearRecordTags(ctx, accountID, groupId)
+
 	return s.requestSGEvent("vpc.delete-sg", SGEvent{
 		GroupId: groupId,
 		VpcId:   record.VpcId,
@@ -34,9 +34,11 @@ type VPCService interface {
 	UpdateSecurityGroupRuleDescriptionsEgress(ctx context.Context, input *ec2.UpdateSecurityGroupRuleDescriptionsEgressInput, accountID string) (*ec2.UpdateSecurityGroupRuleDescriptionsEgressOutput, error)
 }

-// CentralTagWriter projects a create path's record tags into the central tag
-// store so DescribeTags agrees with the record from birth. Implemented by
+// CentralTagStore keeps the central tag index in step with this package's
+// records, so DescribeTags agrees with a resource's own describe for as long as
+// the resource exists and says nothing about it afterwards. Implemented by
 // handlers/ec2/tags.TagsServiceImpl.
-type CentralTagWriter interface {
+type CentralTagStore interface {
 	PutResourceTags(ctx context.Context, accountID, resourceID string, tags map[string]string) error
+	DeleteAllTags(ctx context.Context, accountID, resourceID string) error
 }
@@ -92,9 +92,9 @@ type VPCServiceImpl struct {
 	eipKV        jetstream.KeyValue

 	// Optional: injected after construction so create paths can project their
-	// stored tags into the central tag store. A nil value leaves create
-	// behaviour unchanged.
-	centralTags CentralTagWriter
+	// stored tags into the central tag store and delete paths can clear them.
+	// A nil value leaves both unchanged.
+	centralTags CentralTagStore

 	// disableDefaultPublicIP seeds default subnets with MapPublicIpOnLaunch=false
 	// (non-pool external modes have no public IPs to assign). Zero value keeps
@@ -115,10 +115,11 @@ func (s *VPCServiceImpl) SetExternalIPAM(ipam *ExternalIPAM, eipKV jetstream.Key
 	s.eipKV = eipKV
 }

-// SetCentralTagWriter injects the central tag store writer so create paths
-// project their record tags into it. Nil leaves create behaviour unchanged.
-func (s *VPCServiceImpl) SetCentralTagWriter(w CentralTagWriter) {
-	s.centralTags = w
+// SetCentralTagStore injects the central tag store so create paths project
+// their record tags into it and delete paths clear them. Nil leaves both
+// unchanged.
+func (s *VPCServiceImpl) SetCentralTagStore(st CentralTagStore) {
+	s.centralTags = st
 }

 // localAZ returns the node's local availability zone, sourced from
@@ -494,6 +495,7 @@ func (s *VPCServiceImpl) DeleteVpc(ctx context.Context, input *ec2.DeleteVpcInpu
 		return nil, errors.New(awserrors.ErrorServerInternal)
 	}
 	s.releaseDefaultVPCClaim(ctx, accountID, vpcID)
+	s.clearRecordTags(ctx, accountID, vpcID)

 	slog.InfoContext(ctx, "DeleteVpc completed", "vpcId", vpcID, "accountID", accountID)

@@ -817,6 +819,7 @@ func (s *VPCServiceImpl) DeleteSubnet(ctx context.Context, input *ec2.DeleteSubn
 	if err := s.subnetKV.Delete(ctx, key); err != nil {
 		return nil, errors.New(awserrors.ErrorServerInternal)
 	}
+	s.clearRecordTags(ctx, accountID, subnetID)

 	slog.InfoContext(ctx, "DeleteSubnet completed", "subnetId", subnetID, "accountID", accountID)

@@ -1085,6 +1088,25 @@ func (s *VPCServiceImpl) projectRecordTags(ctx context.Context, accountID, resou
 	}
 }

+// clearRecordTags drops a deleted resource's entry from the central tag store,
+// the delete-direction counterpart to projectRecordTags. On AWS a resource's
+// tags go with it; without this the index answers DescribeTags for resources
+// the caller can no longer describe, and a recycled id inherits them.
+//
+// Call it only once the record delete has succeeded. Clearing first would strip
+// the tags from a resource that then fails to delete and stays live. A failure
+// is logged and swallowed for the same reason the write side swallows one: the
+// resource is already gone, and failing the delete over the index would leave
+// the caller unable to retry something that has happened.
+func (s *VPCServiceImpl) clearRecordTags(ctx context.Context, accountID, resourceID string) {
+	if s.centralTags == nil {
+		return
+	}
+	if err := s.centralTags.DeleteAllTags(ctx, accountID, resourceID); err != nil {
+		slog.ErrorContext(ctx, "central tag store clear failed", "resourceId", resourceID, "err", err)
+	}
+}
+
 // updateRecordTags applies mut to the tag map of the subnet-, vpc-, sg-, or
 // eni-scoped record identified by resourceID. Other resource ids are a no-op.
 func (s *VPCServiceImpl) updateRecordTags(ctx context.Context, accountID, resourceID string, mut func(map[string]string)) error {