fix(aws-rds): version-derived default parameter groups, real DescribeEvents and event-subscription delivery

providers/aws/aws.go

@@ -409,6 +409,14 @@ func wireBedrockAgentRuntime(p *Provider) {
 	p.BedrockAgentRuntime.SetObjectStore(p.S3)
 }
 
+// wireRDS connects RDS to its cross-service dependencies: DB subnet groups
+// resolve their subnets through VPC, and event subscriptions deliver matching
+// RDS events to their SNS topic.
+func wireRDS(p *Provider) {
+	p.RDS.SetSubnetResolver(p.VPC)
+	p.RDS.SetSNSPublisher(p.SNS)
+}
+
 // newProvider builds a provider, using the supplied shared global services when
 // non-nil (multi-region) or a fresh bundle when nil (standalone). The global
 // struct fields are assigned BEFORE the cross-service Set* wiring runs, so every
@@ -450,7 +458,7 @@ func newProvider(o *config.Options, shared *GlobalServices) *Provider {
 	p.Athena.SetMonitoring(p.CloudWatch)
 	p.Athena.SetCatalog(p.Glue)
 	wireBedrockAgentRuntime(p)
-	p.RDS.SetSubnetResolver(p.VPC)
+	wireRDS(p)
 	p.ElastiCache.SetSubnetResolver(p.VPC)
 	p.EC2.SetSubnetResolver(p.VPC)
 	// RunInstances materializes the instance's primary (eth0) ENI in the VPC, and

providers/aws/rds/advanced_restore.go

@@ -108,6 +108,8 @@ func (m *Mock) RestoreDBInstanceToPointInTime(
 		return nil, cerrors.New(cerrors.InvalidArgument, "TargetDBInstanceIdentifier is required")
 	}
 
+	defer m.flushEventNotifications() // runs after the unlock below
+
 	m.mu.Lock()
 	defer m.mu.Unlock()
 
@@ -158,6 +160,8 @@ func (m *Mock) RestoreDBInstanceToPointInTime(
 
 	m.instances.Set(inst.ID, inst)
 	m.rootPasswords[inst.ID] = restoreCfg.MasterUserPassword
+	m.recordEventLocked(eventSource{sourceDBInstance, inst.ID, inst.ARN}, evInstancePITR,
+		input.SourceInstanceID, inst.ID)
 
 	m.emitInstanceMetrics(inst.ID, inst.Engine, cpuMetricRunning, connectionsRunning)
 

providers/aws/rds/aurora.go

@@ -205,6 +205,8 @@ func (m *Mock) DeleteDBClusterEndpoint(_ context.Context, endpointID string) (*r
 // ---- failover ----
 
 func (m *Mock) FailoverDBCluster(_ context.Context, clusterID, targetInstanceID string) (*rdsdriver.Cluster, error) {
+	defer m.flushEventNotifications() // runs after the unlock below
+
 	m.mu.Lock()
 	defer m.mu.Unlock()
 
@@ -236,6 +238,7 @@ func (m *Mock) FailoverDBCluster(_ context.Context, clusterID, targetInstanceID
 	}
 
 	m.clusters.Set(clusterID, cluster)
+	m.recordEventLocked(eventSource{sourceDBCluster, clusterID, cluster.ARN}, evClusterFailover, cluster.Members[0])
 
 	out := cluster
 

providers/aws/rds/dbproxy.go

@@ -36,6 +36,8 @@ func (m *Mock) CreateDBProxy(ctx context.Context, cfg rdsdriver.DBProxyConfig) (
 		return nil, cerrors.New(cerrors.InvalidArgument, "EngineFamily is required")
 	}
 
+	defer m.flushEventNotifications() // runs after the unlock below
+
 	m.mu.Lock()
 	defer m.mu.Unlock()
 
@@ -60,6 +62,7 @@ func (m *Mock) CreateDBProxy(ctx context.Context, cfg rdsdriver.DBProxyConfig) (
 		CreatedAt:           m.opts.Clock.Now().UTC(),
 	}
 	m.proxies.Set(cfg.Name, proxy)
+	m.recordEventLocked(eventSource{sourceDBProxy, cfg.Name, proxy.ARN}, evProxyCreated, cfg.Name)
 
 	out := proxy
 
@@ -108,6 +111,8 @@ func cloneProxy(p rdsdriver.DBProxy) rdsdriver.DBProxy {
 }
 
 func (m *Mock) ModifyDBProxy(_ context.Context, name string, input rdsdriver.ModifyDBProxyInput) (*rdsdriver.DBProxy, error) {
+	defer m.flushEventNotifications() // runs after the unlock below
+
 	m.mu.Lock()
 	defer m.mu.Unlock()
 
@@ -133,13 +138,16 @@ func (m *Mock) ModifyDBProxy(_ context.Context, name string, input rdsdriver.Mod
 	}
 
 	m.proxies.Set(name, p)
+	m.recordEventLocked(eventSource{sourceDBProxy, name, p.ARN}, evProxyModified, name)
 
 	out := p
 
 	return &out, nil
 }
 
 func (m *Mock) DeleteDBProxy(_ context.Context, name string) (*rdsdriver.DBProxy, error) {
+	defer m.flushEventNotifications() // runs after the unlock below
+
 	m.mu.Lock()
 	defer m.mu.Unlock()
 
@@ -151,6 +159,7 @@ func (m *Mock) DeleteDBProxy(_ context.Context, name string) (*rdsdriver.DBProxy
 	p.Status = "deleting"
 
 	m.proxies.Delete(name)
+	m.recordEventLocked(eventSource{sourceDBProxy, name, p.ARN}, evProxyDeleted, name)
 
 	out := p