diff --git a/Cargo.lock b/Cargo.lock index ed5c21271..b62977acc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5618,10 +5618,12 @@ dependencies = [ "fakecloud-core", "fakecloud-k8s", "fakecloud-persistence", + "fakecloud-s3", "form_urlencoded", "http 1.4.0", "k8s-openapi", "kube", + "md-5 0.10.6", "parking_lot", "serde", "serde_json", diff --git a/crates/fakecloud-e2e/tests/elasticache.rs b/crates/fakecloud-e2e/tests/elasticache.rs index d4cec2ea0..e8c975aab 100644 --- a/crates/fakecloud-e2e/tests/elasticache.rs +++ b/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. diff --git a/crates/fakecloud-elasticache/Cargo.toml b/crates/fakecloud-elasticache/Cargo.toml index a5214bb1c..922f82c61 100644 --- a/crates/fakecloud-elasticache/Cargo.toml +++ b/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 } diff --git a/crates/fakecloud-elasticache/src/helpers.rs b/crates/fakecloud-elasticache/src/helpers.rs index 85bda48b0..12960eb56 100644 --- a/crates/fakecloud-elasticache/src/helpers.rs +++ b/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 `-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!( + "\ + {cluster_id}\ + 0001\ + {primary_az_xml}\ + {role}\ + ", + cluster_id = xml_escape(&cluster_id), + ) + }) + .collect(); format!( "\ {shard:04}\ @@ -1223,16 +1252,8 @@ pub(crate) fn replication_group_xml(g: &ReplicationGroup, region: &str) -> Strin
{endpoint_address}
\ {endpoint_port}\ \ - \ - \ - {primary_cluster}\ - 0001\ - {primary_az_xml}\ - primary\ - \ - \ + {members_xml}\ ", - primary_cluster = xml_escape(primary_cluster), ) }) .collect(); diff --git a/crates/fakecloud-elasticache/src/service/mod.rs b/crates/fakecloud-elasticache/src/service/mod.rs index 9f4bbe0e7..5636a1f35 100644 --- a/crates/fakecloud-elasticache/src/service/mod.rs +++ b/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>, snapshot_store: Option>, snapshot_lock: Arc>, + /// 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, } 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) -> Self { self.snapshot_store = Some(store); self diff --git a/crates/fakecloud-elasticache/src/service/serverless.rs b/crates/fakecloud-elasticache/src/service/serverless.rs index 302262915..fdf695b54 100644 --- a/crates/fakecloud-elasticache/src/service/serverless.rs +++ b/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(), diff --git a/crates/fakecloud-elasticache/src/service/snapshots.rs b/crates/fakecloud-elasticache/src/service/snapshots.rs index 0401500c2..a752a588d 100644 --- a/crates/fakecloud-elasticache/src/service/snapshots.rs +++ b/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 { 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)] diff --git a/crates/fakecloud-elasticache/src/service_tests.rs b/crates/fakecloud-elasticache/src/service_tests.rs index 5d4bed194..17ab778f0 100644 --- a/crates/fakecloud-elasticache/src/service_tests.rs +++ b/crates/fakecloud-elasticache/src/service_tests.rs @@ -5015,6 +5015,177 @@ async fn snapshot_hook_fires_with_store() { hook().await; } +#[tokio::test] +async fn create_serverless_cache_without_runtime_assigns_default_port() { + // No container runtime: the cache is created directly "available", so it + // must advertise a real port (6379 for redis/valkey) rather than 0, which + // would make every connection string end in `:0` (unconnectable). + // bug-hunt 2026-07-16, 1.7. + let shared = std::sync::Arc::new(parking_lot::RwLock::new( + fakecloud_core::multi_account::MultiAccountState::new("123456789012", "us-east-1", ""), + )); + let service = ElastiCacheService::new(shared); + + let resp = service + .create_serverless_cache(&request( + "CreateServerlessCache", + &[("ServerlessCacheName", "sc-redis"), ("Engine", "redis")], + )) + .await + .expect("metadata-only create"); + let body = String::from_utf8(resp.body.expect_bytes().to_vec()).unwrap(); + assert!(body.contains("available")); + assert!(body.contains("6379"), "body: {body}"); + assert!( + !body.contains("0"), + "endpoint port must not be 0" + ); + + // Describe round-trips the non-zero port too. + let desc = service + .describe_serverless_caches(&request( + "DescribeServerlessCaches", + &[("ServerlessCacheName", "sc-redis")], + )) + .unwrap(); + let desc_body = String::from_utf8(desc.body.expect_bytes().to_vec()).unwrap(); + assert!(desc_body.contains("6379"), "desc: {desc_body}"); + assert!(!desc_body.contains("0")); +} + +#[tokio::test] +async fn replication_group_emits_primary_and_replica_members() { + // A replication group created with NumCacheClusters=3 has 1 primary + 2 + // replicas. The Terraform provider derives replicas_per_node_group from + // len(NodeGroupMembers) - 1, so all three must be emitted or the plan + // perpetually diffs. bug-hunt 2026-07-16, 1.8. + let shared = std::sync::Arc::new(parking_lot::RwLock::new( + fakecloud_core::multi_account::MultiAccountState::new("123456789012", "us-east-1", ""), + )); + let service = ElastiCacheService::new(shared); + + service + .create_replication_group(&request( + "CreateReplicationGroup", + &[ + ("ReplicationGroupId", "rg-repl"), + ("ReplicationGroupDescription", "test"), + ("NumCacheClusters", "3"), + ], + )) + .await + .expect("create replication group"); + + let resp = service + .describe_replication_groups(&request( + "DescribeReplicationGroups", + &[("ReplicationGroupId", "rg-repl")], + )) + .unwrap(); + let body = String::from_utf8(resp.body.expect_bytes().to_vec()).unwrap(); + assert_eq!( + body.matches("").count(), + 3, + "expected 1 primary + 2 replicas, body: {body}" + ); + assert_eq!( + body.matches("primary").count(), + 1 + ); + assert_eq!( + body.matches("replica").count(), + 2 + ); + assert!(body.contains("rg-repl-001")); + assert!(body.contains("rg-repl-002")); + assert!(body.contains("rg-repl-003")); +} + +#[tokio::test] +async fn export_serverless_cache_snapshot_writes_object_to_s3() { + // ExportServerlessCacheSnapshot must actually drop the export artifact into + // the caller's S3 bucket rather than discard S3BucketName. bug-hunt + // 2026-07-16, 1.25. + use fakecloud_s3::{S3Bucket, SharedS3State}; + let s3: SharedS3State = std::sync::Arc::new(parking_lot::RwLock::new( + fakecloud_core::multi_account::MultiAccountState::new("123456789012", "us-east-1", ""), + )); + { + let mut guard = s3.write(); + let acc = guard.get_or_create("123456789012"); + acc.buckets.insert( + "export-bucket".to_string(), + S3Bucket::new("export-bucket", "us-east-1", "123456789012"), + ); + } + + let shared = std::sync::Arc::new(parking_lot::RwLock::new( + fakecloud_core::multi_account::MultiAccountState::new("123456789012", "us-east-1", ""), + )); + let service = ElastiCacheService::new(shared).with_s3(s3.clone()); + { + let mut __a = service.state.write(); + let state = __a.default_mut(); + state.serverless_cache_snapshots.insert( + "snap-x".to_string(), + ServerlessCacheSnapshot { + serverless_cache_snapshot_name: "snap-x".to_string(), + arn: "arn:aws:elasticache:us-east-1:123456789012:serverlesssnapshot:snap-x" + .to_string(), + kms_key_id: None, + snapshot_type: "manual".to_string(), + status: "available".to_string(), + create_time: "2024-01-01T00:00:00Z".to_string(), + expiry_time: None, + bytes_used_for_cache: None, + serverless_cache_name: "cache-a".to_string(), + engine: "redis".to_string(), + major_engine_version: "7.1".to_string(), + }, + ); + } + + let resp = service + .export_serverless_cache_snapshot(&request( + "ExportServerlessCacheSnapshot", + &[ + ("ServerlessCacheSnapshotName", "snap-x"), + ("S3BucketName", "export-bucket"), + ], + )) + .expect("export succeeds"); + let body = String::from_utf8(resp.body.expect_bytes().to_vec()).unwrap(); + assert!(body.contains("snap-x")); + + // The export object now exists in the target bucket with real bytes + ETag. + { + let guard = s3.read(); + let acc = guard.get("123456789012").expect("s3 account"); + let bucket = acc.buckets.get("export-bucket").expect("bucket"); + let obj = bucket + .objects + .get("snap-x.rdb") + .expect("export object written to S3"); + assert!(obj.size > 0); + assert!(obj.etag.starts_with('"') && obj.etag.ends_with('"')); + } + + // A missing bucket is rejected the way AWS does, not silently dropped. + match service.export_serverless_cache_snapshot(&request( + "ExportServerlessCacheSnapshot", + &[ + ("ServerlessCacheSnapshotName", "snap-x"), + ("S3BucketName", "nonexistent-bucket"), + ], + )) { + Ok(_) => panic!("expected missing-bucket rejection"), + Err(e) => { + assert_eq!(e.status(), StatusCode::BAD_REQUEST); + assert_eq!(e.code(), "InvalidParameterValue"); + } + } +} + #[test] fn recovery_predicate_redrives_transient_states() { use crate::service::is_recoverable_status; diff --git a/crates/fakecloud-server/src/main.rs b/crates/fakecloud-server/src/main.rs index 5dd0dbd09..7d4fd0929 100644 --- a/crates/fakecloud-server/src/main.rs +++ b/crates/fakecloud-server/src/main.rs @@ -3006,7 +3006,8 @@ async fn main() { } else { None }; - let mut elasticache_service = ElastiCacheService::new(elasticache_state); + let mut elasticache_service = + ElastiCacheService::new(elasticache_state).with_s3(s3_state.clone()); if let Some(ref rt) = elasticache_runtime { elasticache_service = elasticache_service.with_runtime(rt.clone()); }