fix(elasticache): serverless endpoint port + replication-group members + snapshot export (bug-hunt)

crates/fakecloud-e2e/tests/elasticache.rs

@@ -953,6 +953,20 @@ async fn elasticache_create_replication_group_round_trips_extended_fields() {
     assert!(ids.contains(&"0001"));
     assert!(ids.contains(&"0002"));
     assert!(ids.contains(&"0003"));
+    // replicas_per_node_group=1 -> each shard exposes 1 primary + 1 replica
+    // NodeGroupMember. The Terraform provider derives replicas_per_node_group
+    // from len(NodeGroupMembers) - 1, so a single member perpetually diffs.
+    for ng in node_groups {
+        let members = ng.node_group_members();
+        assert_eq!(
+            members.len(),
+            2,
+            "each shard must expose 1 primary + 1 replica member"
+        );
+        let roles: Vec<&str> = members.iter().filter_map(|m| m.current_role()).collect();
+        assert!(roles.contains(&"primary"), "roles: {roles:?}");
+        assert!(roles.contains(&"replica"), "roles: {roles:?}");
+    }
     assert_eq!(described.snapshot_retention_limit(), Some(7));
     assert_eq!(described.snapshot_window(), Some("03:00-04:00"));
     // AuthToken is never echoed on describe, even when stored internally.

crates/fakecloud-elasticache/Cargo.toml

@@ -12,10 +12,13 @@ fakecloud-aws = { workspace = true }
 fakecloud-core = { workspace = true }
 fakecloud-k8s = { workspace = true }
 fakecloud-persistence = { workspace = true }
+fakecloud-s3 = { workspace = true }
 k8s-openapi = { workspace = true }
 async-trait = { workspace = true }
+bytes = { workspace = true }
 chrono = { workspace = true }
 http = { workspace = true }
+md-5 = { workspace = true }
 parking_lot = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true }
@@ -25,7 +28,6 @@ serde_json = { workspace = true }
 tracing = { workspace = true }
 
 [dev-dependencies]
-bytes = { workspace = true }
 # The opt-in k8s-integration test drives a real cluster directly.
 kube = { workspace = true }
 

crates/fakecloud-elasticache/src/helpers.rs

@@ -1201,16 +1201,45 @@ pub(crate) fn replication_group_xml(g: &ReplicationGroup, region: &str) -> Strin
     // 500-shard ceiling so a corrupt stored value can't cause an
     // unbounded XML allocation here.
     const MAX_SHARDS: i32 = 500;
+    // AWS caps a Redis/Valkey node group at 5 read replicas. Clamp so a
+    // corrupt stored value can't drive an unbounded member allocation.
+    const MAX_REPLICAS_PER_SHARD: i32 = 5;
     let shard_count = g.num_node_groups.clamp(1, MAX_SHARDS);
+    // Emit one primary NodeGroupMember plus one member per configured replica.
+    // The Terraform provider derives `replicas_per_node_group` from
+    // `len(NodeGroupMembers) - 1`; emitting only the primary reported 0
+    // replicas and produced a perpetual plan diff. bug-hunt 2026-07-16, 1.8.
+    let replicas_per_shard = current_replicas_per_shard(g).clamp(0, MAX_REPLICAS_PER_SHARD);
+    let members_per_shard = replicas_per_shard + 1;
     let node_groups_inner: String = (1..=shard_count)
         .map(|shard| {
-            // Pull the matching member cluster id when possible so multi-shard
-            // describe responses round-trip the requested NumNodeGroups.
-            let primary_cluster = g
-                .member_clusters
-                .get((shard - 1) as usize)
-                .map(|s| s.as_str())
-                .unwrap_or_else(|| g.member_clusters.first().map(|s| s.as_str()).unwrap_or(""));
+            // Members for this shard: 1 primary + N replicas. AWS names the
+            // per-node clusters `<rg>-001` (primary), `-002..` (replicas);
+            // reuse the stored member_clusters slice when it lines up so
+            // multi-shard describe responses round-trip the requested shape.
+            let base = ((shard - 1) * members_per_shard) as usize;
+            let members_xml: String = (0..members_per_shard)
+                .map(|i| {
+                    let global_idx = base + i as usize;
+                    let cluster_id =
+                        g.member_clusters
+                            .get(global_idx)
+                            .cloned()
+                            .unwrap_or_else(|| {
+                                format!("{}-{:03}", g.replication_group_id, global_idx + 1)
+                            });
+                    let role = if i == 0 { "primary" } else { "replica" };
+                    format!(
+                        "<NodeGroupMember>\
+                         <CacheClusterId>{cluster_id}</CacheClusterId>\
+                         <CacheNodeId>0001</CacheNodeId>\
+                         <PreferredAvailabilityZone>{primary_az_xml}</PreferredAvailabilityZone>\
+                         <CurrentRole>{role}</CurrentRole>\
+                         </NodeGroupMember>",
+                        cluster_id = xml_escape(&cluster_id),
+                    )
+                })
+                .collect();
             format!(
                 "<NodeGroup>\
                  <NodeGroupId>{shard:04}</NodeGroupId>\
@@ -1223,16 +1252,8 @@ pub(crate) fn replication_group_xml(g: &ReplicationGroup, region: &str) -> Strin
                  <Address>{endpoint_address}</Address>\
                  <Port>{endpoint_port}</Port>\
                  </ReaderEndpoint>\
-                 <NodeGroupMembers>\
-                 <NodeGroupMember>\
-                 <CacheClusterId>{primary_cluster}</CacheClusterId>\
-                 <CacheNodeId>0001</CacheNodeId>\
-                 <PreferredAvailabilityZone>{primary_az_xml}</PreferredAvailabilityZone>\
-                 <CurrentRole>primary</CurrentRole>\
-                 </NodeGroupMember>\
-                 </NodeGroupMembers>\
+                 <NodeGroupMembers>{members_xml}</NodeGroupMembers>\
                  </NodeGroup>",
-                primary_cluster = xml_escape(primary_cluster),
             )
         })
         .collect();

crates/fakecloud-elasticache/src/service/mod.rs

@@ -9,6 +9,7 @@ use fakecloud_aws::xml::xml_escape;
 use fakecloud_core::query::{optional_query_param, query_response_xml, required_query_param};
 use fakecloud_core::service::{AwsRequest, AwsResponse, AwsService, AwsServiceError};
 use fakecloud_persistence::SnapshotStore;
+use fakecloud_s3::{memory_body, S3Object, SharedS3State};
 
 use crate::runtime::ElastiCacheRuntime;
 use crate::state::{
@@ -172,6 +173,10 @@ pub struct ElastiCacheService {
     runtime: Option<Arc<ElastiCacheRuntime>>,
     snapshot_store: Option<Arc<dyn SnapshotStore>>,
     snapshot_lock: Arc<AsyncMutex<()>>,
+    /// Shared S3 state so `ExportServerlessCacheSnapshot` can write the
+    /// exported artifact into the target bucket, matching AWS. `None` in
+    /// unit tests / when S3 isn't wired.
+    s3: Option<SharedS3State>,
 }
 
 mod clusters;
@@ -191,6 +196,7 @@ impl ElastiCacheService {
             runtime: None,
             snapshot_store: None,
             snapshot_lock: Arc::new(AsyncMutex::new(())),
+            s3: None,
         }
     }
 
@@ -199,6 +205,13 @@ impl ElastiCacheService {
         self
     }
 
+    /// Wire the shared S3 state used by `ExportServerlessCacheSnapshot` to
+    /// write the exported snapshot artifact into the target bucket.
+    pub fn with_s3(mut self, s3: SharedS3State) -> Self {
+        self.s3 = Some(s3);
+        self
+    }
+
     pub fn with_snapshot_store(mut self, store: Arc<dyn SnapshotStore>) -> Self {
         self.snapshot_store = Some(store);
         self

crates/fakecloud-elasticache/src/service/serverless.rs

@@ -67,17 +67,30 @@ impl ElastiCacheService {
             (arn, "127.0.0.1".to_string())
         };
 
-        // The backing container is started off the request path (below); the
-        // cache begins with no endpoint port and is filled in once ready, so the
-        // response doesn't block on the cold image pull + readiness (which made
-        // the AWS CLI hit its 60s read timeout). bug-audit 2026-05-28, 3.2.
+        // The backing container is started off the request path (below); with a
+        // runtime the cache begins with no endpoint port and is filled in once
+        // ready, so the response doesn't block on the cold image pull +
+        // readiness (which made the AWS CLI hit its 60s read timeout).
+        // bug-audit 2026-05-28, 3.2.
+        //
+        // Without a runtime (the default in local/CI setups) the cache is
+        // created directly as "available", so it must advertise a real port
+        // immediately — the Docker-tick backfill never runs, so leaving it 0
+        // makes every connection string end in `:0` (unconnectable). Use the
+        // conventional engine port (6379 redis/valkey, 11211 memcached).
+        // bug-hunt 2026-07-16, 1.7.
+        let initial_port = if self.runtime.is_some() {
+            0
+        } else {
+            engine_default_port(&engine)
+        };
         let endpoint = ServerlessCacheEndpoint {
             address: endpoint_address.clone(),
-            port: 0,
+            port: initial_port,
         };
         let reader_endpoint = ServerlessCacheEndpoint {
             address: endpoint_address,
-            port: 0,
+            port: initial_port,
         };
         let cache = ServerlessCache {
             serverless_cache_name: serverless_cache_name.clone(),

crates/fakecloud-elasticache/src/service/snapshots.rs

@@ -1,6 +1,9 @@
 //! ElastiCache `snapshots` family handlers extracted from service.rs
 //! by audit-2026-05-19 file-split.
 
+use bytes::Bytes;
+use md5::{Digest, Md5};
+
 use super::*;
 
 /// Per-snapshot RDB dump path. Roots under the platform temp dir rather than a
@@ -540,21 +543,31 @@ impl ElastiCacheService {
     ) -> Result<AwsResponse, AwsServiceError> {
         let snap_name = required_query_param(request, "ServerlessCacheSnapshotName")?;
         let bucket = required_query_param(request, "S3BucketName")?;
-        let accounts = self.state.read();
-        let empty = ElastiCacheState::new(&request.account_id, &request.region);
-        let state = accounts.get(&request.account_id).unwrap_or(&empty);
-        let snap = state
-            .serverless_cache_snapshots
-            .get(&snap_name)
-            .ok_or_else(|| {
-                AwsServiceError::aws_error(
-                    StatusCode::NOT_FOUND,
-                    "ServerlessCacheSnapshotNotFoundFault",
-                    format!("ServerlessCacheSnapshot {snap_name} not found."),
-                )
-            })?;
-        let xml = serverless_cache_snapshot_xml(snap);
-        let _ = bucket;
+        let snap = {
+            let accounts = self.state.read();
+            let empty = ElastiCacheState::new(&request.account_id, &request.region);
+            let state = accounts.get(&request.account_id).unwrap_or(&empty);
+            state
+                .serverless_cache_snapshots
+                .get(&snap_name)
+                .cloned()
+                .ok_or_else(|| {
+                    AwsServiceError::aws_error(
+                        StatusCode::NOT_FOUND,
+                        "ServerlessCacheSnapshotNotFoundFault",
+                        format!("ServerlessCacheSnapshot {snap_name} not found."),
+                    )
+                })?
+        };
+
+        // Actually perform the export: AWS drops the snapshot's RDB artifact
+        // into the caller's S3 bucket. We synthesize a small artifact carrying
+        // the snapshot metadata and write it through the shared S3 state so a
+        // follow-up GetObject returns it. The bucket must already exist in this
+        // account (AWS rejects otherwise). bug-hunt 2026-07-16, 1.25.
+        self.export_snapshot_artifact_to_s3(&request.account_id, &bucket, &snap)?;
+
+        let xml = serverless_cache_snapshot_xml(&snap);
         Ok(AwsResponse::xml(
             StatusCode::OK,
             query_response_xml(
@@ -565,6 +578,59 @@ impl ElastiCacheService {
             ),
         ))
     }
+
+    /// Write the serverless snapshot export artifact into the target S3 bucket.
+    /// Returns `InvalidParameterValue` when the bucket does not exist in the
+    /// account (matching AWS), and is a no-op when S3 isn't wired (unit tests).
+    fn export_snapshot_artifact_to_s3(
+        &self,
+        account_id: &str,
+        bucket: &str,
+        snap: &ServerlessCacheSnapshot,
+    ) -> Result<(), AwsServiceError> {
+        let Some(ref s3) = self.s3 else {
+            return Ok(());
+        };
+        let object_key = format!("{}.rdb", snap.serverless_cache_snapshot_name);
+        let body_bytes = serde_json::json!({
+            "serverlessCacheSnapshotName": snap.serverless_cache_snapshot_name,
+            "serverlessCacheName": snap.serverless_cache_name,
+            "arn": snap.arn,
+            "engine": snap.engine,
+            "majorEngineVersion": snap.major_engine_version,
+            "snapshotType": snap.snapshot_type,
+            "createTime": snap.create_time,
+        })
+        .to_string()
+        .into_bytes();
+
+        let mut hasher = Md5::new();
+        hasher.update(&body_bytes);
+        let etag = format!("\"{:x}\"", hasher.finalize());
+        let size = body_bytes.len() as u64;
+
+        let mut s3_state = s3.write();
+        let s3_account = s3_state.get_or_create(account_id);
+        let bucket_state = s3_account.buckets.get_mut(bucket).ok_or_else(|| {
+            AwsServiceError::aws_error(
+                StatusCode::BAD_REQUEST,
+                "InvalidParameterValue",
+                format!("The S3 bucket {bucket} does not exist or is not owned by you."),
+            )
+        })?;
+        let object = S3Object {
+            key: object_key.clone(),
+            body: memory_body(Bytes::from(body_bytes)),
+            content_type: "application/octet-stream".to_string(),
+            etag,
+            size,
+            last_modified: chrono::Utc::now(),
+            storage_class: "STANDARD".to_string(),
+            ..Default::default()
+        };
+        bucket_state.objects.insert(object_key, object);
+        Ok(())
+    }
 }
 
 #[cfg(test)]