fix(aws): skip resources deleted mid-walk in every RGT discovery walker

providers/aws/rds_discovery.go

@@ -133,6 +133,13 @@ func (a rdsDiscovery) discoverRedshift(ctx context.Context) ([]resourcediscovery
 		c := clusters[i]
 
 		tags, err := a.redshift.DescribeTags(ctx, c.ARN)
+		// NotFound means the cluster was deleted between DescribeClusters and
+		// this tag read. Skip it, the same as one that was never listed. Any
+		// other error propagates.
+		if cerrors.IsNotFound(err) {
+			continue
+		}
+
 		if err != nil {
 			return nil, err
 		}

providers/aws/rgt_discovery.go

@@ -2,6 +2,7 @@ package aws
 
 import (
 	"context"
+	"slices"
 
 	cerrors "github.com/stackshy/cloudemu/v2/errors"
 	"github.com/stackshy/cloudemu/v2/internal/idgen"
@@ -14,9 +15,11 @@ import (
 	guarddutydriver "github.com/stackshy/cloudemu/v2/services/guardduty/driver"
 	kafkadriver "github.com/stackshy/cloudemu/v2/services/kafka/driver"
 	ksdriver "github.com/stackshy/cloudemu/v2/services/keyspaces/driver"
+	kinesisdriver "github.com/stackshy/cloudemu/v2/services/kinesis/driver"
 	kmsdriver "github.com/stackshy/cloudemu/v2/services/kms/driver"
 	mdbdriver "github.com/stackshy/cloudemu/v2/services/memorydb/driver"
 	nfdriver "github.com/stackshy/cloudemu/v2/services/networkfirewall/driver"
+	opensearchdriver "github.com/stackshy/cloudemu/v2/services/opensearch/driver"
 	ssmdriver "github.com/stackshy/cloudemu/v2/services/parameterstore/driver"
 	"github.com/stackshy/cloudemu/v2/services/resourcediscovery"
 	r53rdriver "github.com/stackshy/cloudemu/v2/services/route53resolver/driver"
@@ -85,21 +88,14 @@ func (d taggedDiscovery) discoverOpenSearch(ctx context.Context) ([]resourcedisc
 		return nil, err
 	}
 
-	out := make([]resourcediscovery.DiscoveredResource, 0, len(domains))
-
-	for i := range domains {
-		name := domains[i].DomainName
-		arn := idgen.AWSARN("es", d.p.Region, d.p.AccountID, "domain/"+name)
-
-		tags, err := d.p.OpenSearch.ListTags(ctx, arn)
-		if err != nil {
-			return nil, err
-		}
+	return project(ctx, domains, d.openSearchRow)
+}
 
-		out = append(out, row("es", "domain", name, arn, tags))
-	}
+func (d taggedDiscovery) openSearchRow(ctx context.Context, dom opensearchdriver.DomainInfo) (resourcediscovery.DiscoveredResource, error) {
+	arn := idgen.AWSARN("es", d.p.Region, d.p.AccountID, "domain/"+dom.DomainName)
+	tags, err := d.p.OpenSearch.ListTags(ctx, arn)
 
-	return out, nil
+	return row("es", "domain", dom.DomainName, arn, tags), err
 }
 
 func (d taggedDiscovery) discoverSFN(ctx context.Context) ([]resourcediscovery.DiscoveredResource, error) {
@@ -272,124 +268,111 @@ func (d taggedDiscovery) discoverWAFv2(ctx context.Context) ([]resourcediscovery
 }
 
 func (d taggedDiscovery) discoverGlue(ctx context.Context) ([]resourcediscovery.DiscoveredResource, error) {
-	var (
-		out   []resourcediscovery.DiscoveredResource
-		token string
-	)
-
-	for {
-		names, next, err := d.p.Glue.ListJobs(ctx, gluedriver.TablePagination{NextToken: token, MaxResults: pageLimit})
-		if err != nil {
-			return nil, err
-		}
-
-		for _, name := range names {
-			arn := idgen.AWSARN("glue", d.p.Region, d.p.AccountID, "job/"+name)
+	names, err := listAll(func(token string) ([]string, string, error) {
+		return d.p.Glue.ListJobs(ctx, gluedriver.TablePagination{NextToken: token, MaxResults: pageLimit})
+	})
+	if err != nil {
+		return nil, err
+	}
 
-			tags, err := d.p.Glue.GetTags(ctx, arn)
-			if err != nil {
-				return nil, err
-			}
+	return project(ctx, names, d.glueJobRow)
+}
 
-			out = append(out, row("glue", "job", name, arn, tags))
-		}
+// glueJobRow reads a Glue job's tags. GetTags reads an ARN-keyed store and
+// returns no error for a job that no longer exists, so GetJob runs first: a job
+// deleted after ListJobs then reports NotFound and is skipped, instead of
+// surfacing as a row for a resource that is gone.
+func (d taggedDiscovery) glueJobRow(ctx context.Context, name string) (resourcediscovery.DiscoveredResource, error) {
+	if _, err := d.p.Glue.GetJob(ctx, name); err != nil {
+		return resourcediscovery.DiscoveredResource{}, err
+	}
 
-		if next == "" {
-			return out, nil
-		}
+	arn := idgen.AWSARN("glue", d.p.Region, d.p.AccountID, "job/"+name)
+	tags, err := d.p.Glue.GetTags(ctx, arn)
 
-		token = next
-	}
+	return row("glue", "job", name, arn, tags), err
 }
 
 func (d taggedDiscovery) discoverKafka(ctx context.Context) ([]resourcediscovery.DiscoveredResource, error) {
-	var (
-		out   []resourcediscovery.DiscoveredResource
-		token string
-	)
-
-	for {
-		clusters, next, err := d.p.Kafka.ListClusters(ctx, "", kafkadriver.Page{NextToken: token, MaxResults: pageLimit})
-		if err != nil {
-			return nil, err
-		}
-
-		for i := range clusters {
-			tags, err := d.p.Kafka.ListTagsForResource(ctx, clusters[i].ClusterARN)
-			if err != nil {
-				return nil, err
-			}
+	clusters, err := listAll(func(token string) ([]kafkadriver.Cluster, string, error) {
+		return d.p.Kafka.ListClusters(ctx, "", kafkadriver.Page{NextToken: token, MaxResults: pageLimit})
+	})
+	if err != nil {
+		return nil, err
+	}
 
-			out = append(out, row("kafka", "cluster", clusters[i].ClusterName, clusters[i].ClusterARN, tags))
-		}
+	return project(ctx, clusters, d.kafkaRow)
+}
 
-		if next == "" {
-			return out, nil
-		}
+//nolint:gocritic // hugeParam: the signature is project's by-value per-item callback.
+func (d taggedDiscovery) kafkaRow(ctx context.Context, c kafkadriver.Cluster) (resourcediscovery.DiscoveredResource, error) {
+	tags, err := d.p.Kafka.ListTagsForResource(ctx, c.ClusterARN)
 
-		token = next
-	}
+	return row("kafka", "cluster", c.ClusterName, c.ClusterARN, tags), err
 }
 
 func (d taggedDiscovery) discoverKinesis(ctx context.Context) ([]resourcediscovery.DiscoveredResource, error) {
-	var (
-		out   []resourcediscovery.DiscoveredResource
-		token string
-	)
-
-	for {
+	streams, err := listAll(func(token string) ([]kinesisdriver.StreamSummary, string, error) {
 		res, err := d.p.Kinesis.ListStreams(ctx, token, "", pageLimit)
 		if err != nil {
-			return nil, err
+			return nil, "", err
 		}
 
-		for i := range res.StreamSummaries {
-			s := res.StreamSummaries[i]
+		if !res.HasMoreStreams {
+			return res.StreamSummaries, "", nil
+		}
 
-			tags, err := d.p.Kinesis.ListTagsForResource(ctx, s.StreamARN)
-			if err != nil {
-				return nil, err
-			}
+		return res.StreamSummaries, res.NextToken, nil
+	})
+	if err != nil {
+		return nil, err
+	}
 
-			out = append(out, row("kinesis", "stream", s.StreamName, s.StreamARN, tags))
-		}
+	return project(ctx, streams, d.kinesisRow)
+}
 
-		if !res.HasMoreStreams || res.NextToken == "" {
-			return out, nil
-		}
+//nolint:gocritic // hugeParam: the signature is project's by-value per-item callback.
+func (d taggedDiscovery) kinesisRow(ctx context.Context, s kinesisdriver.StreamSummary) (resourcediscovery.DiscoveredResource, error) {
+	tags, err := d.p.Kinesis.ListTagsForResource(ctx, s.StreamARN)
 
-		token = res.NextToken
-	}
+	return row("kinesis", "stream", s.StreamName, s.StreamARN, tags), err
 }
 
 func (d taggedDiscovery) discoverGuardDuty(ctx context.Context) ([]resourcediscovery.DiscoveredResource, error) {
-	var (
-		out   []resourcediscovery.DiscoveredResource
-		token string
-	)
-
-	for {
-		ids, next, err := d.p.GuardDuty.ListDetectors(ctx, guarddutydriver.Page{NextToken: token, MaxResults: pageLimit})
-		if err != nil {
-			return nil, err
-		}
+	ids, err := d.listGuardDutyDetectors(ctx)
+	if err != nil {
+		return nil, err
+	}
 
-		for _, id := range ids {
-			det, err := d.p.GuardDuty.GetDetector(ctx, id)
-			if err != nil {
-				return nil, err
-			}
+	return project(ctx, ids, d.guardDutyRow)
+}
 
-			arn := idgen.AWSARN("guardduty", d.p.Region, d.p.AccountID, "detector/"+id)
-			out = append(out, row("guardduty", "detector", id, arn, det.Tags))
-		}
+func (d taggedDiscovery) listGuardDutyDetectors(ctx context.Context) ([]string, error) {
+	return listAll(func(token string) ([]string, string, error) {
+		return d.p.GuardDuty.ListDetectors(ctx, guarddutydriver.Page{NextToken: token, MaxResults: pageLimit})
+	})
+}
 
-		if next == "" {
-			return out, nil
+// guardDutyRow reads a detector's tags through GetDetector. Like real GuardDuty,
+// GetDetector reports a detector that does not exist as BadRequestException,
+// not NotFound, so the error alone can't tell a deleted detector from a real
+// failure. When GetDetector fails, a fresh ListDetectors decides: if the
+// detector is no longer listed it was deleted after the walk listed it, and the
+// row is reported as NotFound so project skips it. Otherwise the original error
+// propagates.
+func (d taggedDiscovery) guardDutyRow(ctx context.Context, id string) (resourcediscovery.DiscoveredResource, error) {
+	det, err := d.p.GuardDuty.GetDetector(ctx, id)
+	if err != nil {
+		if ids, lerr := d.listGuardDutyDetectors(ctx); lerr == nil && !slices.Contains(ids, id) {
+			return resourcediscovery.DiscoveredResource{}, cerrors.Newf(cerrors.NotFound, "detector %s was deleted", id)
 		}
 
-		token = next
+		return resourcediscovery.DiscoveredResource{}, err
 	}
+
+	arn := idgen.AWSARN("guardduty", d.p.Region, d.p.AccountID, "detector/"+id)
+
+	return row("guardduty", "detector", id, arn, det.Tags), nil
 }
 
 func (d taggedDiscovery) discoverConfig(ctx context.Context) ([]resourcediscovery.DiscoveredResource, error) {
@@ -447,6 +430,30 @@ func project[T any](ctx context.Context, items []T,
 	return out, nil
 }
 
+// listAll follows a NextToken-paginated list call to its last page and returns
+// every item, so a paginated walker can hand the full list to project.
+func listAll[T any](list func(token string) ([]T, string, error)) ([]T, error) {
+	var (
+		all   []T
+		token string
+	)
+
+	for {
+		items, next, err := list(token)
+		if err != nil {
+			return nil, err
+		}
+
+		all = append(all, items...)
+
+		if next == "" {
+			return all, nil
+		}
+
+		token = next
+	}
+}
+
 // row builds a DiscoveredResource, dropping an empty tag map so a resource with
 // no tags surfaces with nil Tags (matching the other walkers).
 func row(service, typ, id, arn string, tags map[string]string) resourcediscovery.DiscoveredResource {