diff --git a/crates/fakecloud-autoscaling/src/cfn_provision.rs b/crates/fakecloud-autoscaling/src/cfn_provision.rs index 76326f9c0..a04997666 100644 --- a/crates/fakecloud-autoscaling/src/cfn_provision.rs +++ b/crates/fakecloud-autoscaling/src/cfn_provision.rs @@ -12,9 +12,21 @@ use std::sync::Arc; +use fakecloud_persistence::SnapshotHook; + use crate::AutoScalingService; use crate::SharedAutoScalingState; +/// Snapshot hooks the detached CFN capacity reconcile fires after it mutates +/// shared state, so the launched ASG instances (`autoscaling`) and their +/// backing EC2 records (`ec2`) survive a restart. `None` fields make the +/// corresponding persist a no-op (memory mode / unit tests). +#[derive(Clone, Default)] +pub struct CfnReconcilePersistHooks { + pub autoscaling: Option, + pub ec2: Option, +} + /// Reconcile a CFN-provisioned Auto Scaling Group to its desired capacity by /// launching REAL EC2 instances. No-op if the group is gone (e.g. the stack was /// deleted before reconciliation ran). Intended to be `tokio::spawn`ed by the @@ -29,9 +41,21 @@ pub async fn cfn_reconcile_capacity( group_name: String, account_id: String, region: String, + persist: CfnReconcilePersistHooks, ) { - let svc = AutoScalingService::new(asg_state).with_ec2(ec2_state, ec2_runtime); + // The EC2 hook is set on the service so `apply_capacity` persists the REAL + // EC2 instances it launches; the autoscaling hook is fired below to persist + // the group's `instances` list. Both are required because this reconcile + // runs DETACHED after `CreateStack`/`UpdateStack` has already serialized the + // group with `instances=[]`; without them the launched instances (and their + // EC2 records) vanish on restart (bug-hunt restart-dataloss). + let svc = AutoScalingService::new(asg_state) + .with_ec2(ec2_state, ec2_runtime) + .with_ec2_snapshot_hook(persist.ec2); svc.reconcile_group(&account_id, &group_name, ®ion).await; + if let Some(hook) = persist.autoscaling { + hook().await; + } } /// Terminate the REAL EC2 instances launched for a CFN-provisioned Auto Scaling @@ -54,3 +78,146 @@ pub async fn cfn_terminate_instances( svc.cfn_terminate_instances(&account_id, ®ion, &instance_ids) .await; } + +#[cfg(test)] +mod tests { + use super::*; + use crate::state::{AutoScalingAccounts, AutoScalingGroup, AutoScalingSnapshot}; + use fakecloud_persistence::SnapshotStore; + use parking_lot::{Mutex, RwLock}; + + /// Shared cell holding the bytes a capturing snapshot store last saved. + type Captured = Arc>>>; + + // A `SnapshotStore` that captures the last-saved bytes in memory so a test + // can assert what the persist hook serialized (no tempfile needed). + struct CapturingStore(Captured); + + impl SnapshotStore for CapturingStore { + fn load(&self) -> std::io::Result>> { + Ok(self.0.lock().clone()) + } + fn save(&self, bytes: &[u8]) -> std::io::Result<()> { + *self.0.lock() = Some(bytes.to_vec()); + Ok(()) + } + } + + fn capturing() -> (Captured, Arc) { + let cell: Captured = Arc::new(Mutex::new(None)); + (cell.clone(), Arc::new(CapturingStore(cell))) + } + + fn group_desired(name: &str, desired: i64) -> AutoScalingGroup { + AutoScalingGroup { + name: name.to_string(), + arn: format!( + "arn:aws:autoscaling:us-east-1:123456789012:autoScalingGroup:x:autoScalingGroupName/{name}" + ), + launch_configuration_name: None, + launch_template: None, + min_size: 0, + max_size: desired, + desired_capacity: desired, + default_cooldown: 300, + availability_zones: vec!["us-east-1a".to_string()], + vpc_zone_identifier: None, + health_check_type: "EC2".to_string(), + health_check_grace_period: 0, + target_group_arns: Vec::new(), + load_balancer_names: Vec::new(), + new_instances_protected_from_scale_in: false, + created_time: chrono::Utc::now(), + instances: Vec::new(), + tags: Vec::new(), + status: None, + service_linked_role_arn: String::new(), + } + } + + // bug-hunt restart-dataloss #1/#2: a CFN-provisioned ASG is inserted with + // `instances=[]` and reconciled to capacity by a DETACHED task. That task + // must persist BOTH the launched ASG instances (autoscaling snapshot) and + // their backing EC2 records (EC2 snapshot); otherwise the group reloads + // empty and the EC2 containers leak with no owning state on restart. + #[tokio::test] + async fn cfn_reconcile_persists_asg_and_ec2_instances() { + let account = "123456789012"; + let asg_state: SharedAutoScalingState = Arc::new(RwLock::new(AutoScalingAccounts::new())); + asg_state + .write() + .get_or_create(account) + .groups + .insert("g".to_string(), group_desired("g", 2)); + + let ec2_state: fakecloud_ec2::SharedEc2State = Arc::new(RwLock::new( + fakecloud_core::multi_account::MultiAccountState::new(account, "us-east-1", ""), + )); + + // Persist hooks over capturing stores, built exactly as the server wires + // the autoscaling + EC2 snapshot hooks. + let (asg_cell, asg_store) = capturing(); + let asg_hook = AutoScalingService::new(asg_state.clone()) + .with_snapshot_store(asg_store) + .snapshot_hook(); + assert!(asg_hook.is_some(), "autoscaling hook must be built"); + + let (ec2_cell, ec2_store) = capturing(); + let ec2_hook = fakecloud_ec2::Ec2Service::with_state(ec2_state.clone()) + .with_snapshot_store(ec2_store) + .snapshot_hook(); + assert!(ec2_hook.is_some(), "ec2 hook must be built"); + + // No EC2 runtime -> metadata-only instances (CI path), still real EC2 + // records in `ec2_state`. + cfn_reconcile_capacity( + asg_state.clone(), + ec2_state.clone(), + None, + "g".to_string(), + account.to_string(), + "us-east-1".to_string(), + CfnReconcilePersistHooks { + autoscaling: asg_hook, + ec2: ec2_hook, + }, + ) + .await; + + // In-memory: the group reached desired capacity. + assert_eq!( + asg_state.read().accounts[account].groups["g"] + .instances + .len(), + 2, + "reconcile must launch 2 instances" + ); + + // #1: the autoscaling snapshot the detached reconcile fired contains the + // launched instances (not the empty list CreateStack serialized). + let asg_bytes = asg_cell + .lock() + .clone() + .expect("autoscaling snapshot written"); + let asg_snap: AutoScalingSnapshot = serde_json::from_slice(&asg_bytes).unwrap(); + let persisted = asg_snap.accounts.expect("multi-account snapshot"); + assert_eq!( + persisted.accounts[account].groups["g"].instances.len(), + 2, + "launched ASG instances must survive restart" + ); + + // #2: the EC2 snapshot the reconcile fired contains the backing records, + // so EC2 boot-recovery has a row to re-drive (no container leak). + let ec2_bytes = ec2_cell.lock().clone().expect("ec2 snapshot written"); + let ec2_snap: fakecloud_ec2::Ec2Snapshot = serde_json::from_slice(&ec2_bytes).unwrap(); + let ec2_accounts = ec2_snap.accounts.expect("ec2 multi-account snapshot"); + let in_memory = ec2_state.read().default_ref().instances.len(); + assert_eq!(in_memory, 2, "reconcile must create 2 EC2 records"); + assert_eq!( + ec2_accounts.default_ref().instances.len(), + in_memory, + "ASG-launched EC2 records must be persisted (bug #2)" + ); + } +} diff --git a/crates/fakecloud-autoscaling/src/service.rs b/crates/fakecloud-autoscaling/src/service.rs index c02d70b23..c0a2c899f 100644 --- a/crates/fakecloud-autoscaling/src/service.rs +++ b/crates/fakecloud-autoscaling/src/service.rs @@ -45,6 +45,14 @@ pub struct AutoScalingService { /// instances (unit tests). ec2_state: Option, ec2_runtime: Option>, + /// EC2 whole-state snapshot hook. Reconciling desired capacity launches + /// (and terminates) REAL EC2 instances through a bare `Ec2Service` built + /// without a snapshot store, so those EC2 records lived only in memory and + /// leaked their containers on restart (EC2 boot-recovery had no persisted + /// `pending`/`running` row to re-drive). Firing this hook after an EC2 + /// mutation writes the EC2 state through, the same way a direct + /// `RunInstances` API call persists (bug-hunt restart-dataloss). + ec2_snapshot_hook: Option, } impl AutoScalingService { @@ -55,6 +63,7 @@ impl AutoScalingService { snapshot_lock: Arc::new(AsyncMutex::new(())), ec2_state: None, ec2_runtime: None, + ec2_snapshot_hook: None, } } @@ -63,6 +72,16 @@ impl AutoScalingService { self } + /// Attach the EC2 snapshot hook so ASG-launched EC2 instances are persisted + /// after capacity reconciliation. Without it, ASG-launched instances live + /// only in memory and their backing containers leak on restart (there is no + /// persisted EC2 row for boot-recovery to re-drive). `None` (memory mode / + /// unit tests) makes the persist a no-op. + pub fn with_ec2_snapshot_hook(mut self, hook: Option) -> Self { + self.ec2_snapshot_hook = hook; + self + } + /// Attach the EC2 backend so desired-capacity reconciliation launches real /// container-backed instances via `RunInstances`. pub fn with_ec2( @@ -995,12 +1014,18 @@ impl AutoScalingService { let mut launched: Vec = Vec::new(); let mut terminate_ids: Vec = Vec::new(); + // Whether this reconcile mutated REAL EC2 state (launched or terminated + // container-backed instances), so it must be persisted through the EC2 + // snapshot hook below. No-op when there is no EC2 backend (the ids are + // synthesized metadata) or no hook wired. + let mut ec2_touched = false; if current_ids.len() < target { let need = target - current_ids.len(); let mut ids = self .run_ec2_instances(&image_id, &instance_type, subnet.as_deref(), need, req) .await; + ec2_touched = self.ec2_state.is_some(); // No EC2 backend wired (unit tests) or a partial launch: synthesize // the remainder so the group still reports its desired capacity. while ids.len() < need { @@ -1024,34 +1049,51 @@ impl AutoScalingService { let remove = current_ids.len() - target; terminate_ids = current_ids.iter().rev().take(remove).cloned().collect(); self.terminate_ec2_instances(&terminate_ids, req).await; + ec2_touched = self.ec2_state.is_some(); } - let mut accounts = self.state.write(); - let st = accounts.get_or_create(account); - let descs: Vec = { - let Some(g) = st.groups.get_mut(name) else { - return; - }; - let lcn = g.launch_configuration_name.clone(); - let prot = g.new_instances_protected_from_scale_in; - let mut descs = Vec::new(); - for mut ni in launched { - ni.launch_configuration_name = lcn.clone(); - ni.protected_from_scale_in = prot; - descs.push(format!("Launching a new EC2 instance: {}", ni.instance_id)); - g.instances.push(ni); + // Persist the REAL EC2 records this reconcile launched/terminated BEFORE + // touching the ASG state below. The ASG-launched instances are driven + // through a bare `Ec2Service` built without a snapshot store, so without + // firing the EC2 snapshot hook here they live only in memory and leak + // their containers on restart (EC2 boot-recovery has no persisted row to + // re-drive). Done before the ASG write so a concurrent group delete (the + // early `return` below) cannot skip persisting them. No-op when there is + // no EC2 backend or no hook wired (bug-hunt restart-dataloss). + if ec2_touched { + if let Some(hook) = &self.ec2_snapshot_hook { + hook().await; } - if !terminate_ids.is_empty() { - g.instances - .retain(|i| !terminate_ids.contains(&i.instance_id)); - for id in &terminate_ids { - descs.push(format!("Terminating EC2 instance: {id}")); + } + + { + let mut accounts = self.state.write(); + let st = accounts.get_or_create(account); + let descs: Vec = { + let Some(g) = st.groups.get_mut(name) else { + return; + }; + let lcn = g.launch_configuration_name.clone(); + let prot = g.new_instances_protected_from_scale_in; + let mut descs = Vec::new(); + for mut ni in launched { + ni.launch_configuration_name = lcn.clone(); + ni.protected_from_scale_in = prot; + descs.push(format!("Launching a new EC2 instance: {}", ni.instance_id)); + g.instances.push(ni); } + if !terminate_ids.is_empty() { + g.instances + .retain(|i| !terminate_ids.contains(&i.instance_id)); + for id in &terminate_ids { + descs.push(format!("Terminating EC2 instance: {id}")); + } + } + descs + }; + for d in descs { + st.activities.insert(0, activity(name, &d)); } - descs - }; - for d in descs { - st.activities.insert(0, activity(name, &d)); } } } diff --git a/crates/fakecloud-cloudformation/src/extras.rs b/crates/fakecloud-cloudformation/src/extras.rs index 88e28ae2c..ff324d6ae 100644 --- a/crates/fakecloud-cloudformation/src/extras.rs +++ b/crates/fakecloud-cloudformation/src/extras.rs @@ -1075,7 +1075,8 @@ impl CloudFormationService { // custom-resource Lambda invokes (0.2). { let handles = - crate::service::ContainerBackingHandles::from_provisioner(&provisioner); + crate::service::ContainerBackingHandles::from_provisioner(&provisioner) + .with_snapshot_hooks(&self.snapshot_hooks); handles.spawn_container_intents(std::mem::take( &mut *provisioner.pending_container_spawns.lock(), )); diff --git a/crates/fakecloud-cloudformation/src/service.rs b/crates/fakecloud-cloudformation/src/service.rs index 0bb61e57c..414496925 100644 --- a/crates/fakecloud-cloudformation/src/service.rs +++ b/crates/fakecloud-cloudformation/src/service.rs @@ -494,7 +494,7 @@ pub struct CloudFormationService { /// is written through to disk, the same persistence a direct mutating API /// call would trigger. Empty by default (memory mode, or no services /// wired); the server populates it via `with_snapshot_hooks`. - snapshot_hooks: BTreeMap<&'static str, SnapshotHook>, + pub(crate) snapshot_hooks: BTreeMap<&'static str, SnapshotHook>, } /// Everything the async CreateStack provisioning task needs to provision @@ -541,6 +541,15 @@ pub(crate) struct ContainerBackingHandles { mq_runtime: Option>, kafka_state: fakecloud_kafka::SharedKafkaState, kafka_runtime: Option>, + /// Snapshot hooks for the services whose DETACHED container-backing task + /// mutates snapshot-backed state AFTER the stack op has already serialized + /// it. `persist_touched_services` fires right after the op returns, which + /// races (and usually loses to) the still-running background reconcile, so + /// an Auto Scaling Group's launched `instances` (and their EC2 records) were + /// serialized empty and vanished on restart. The background task fires these + /// hooks itself once it has finished mutating (bug-hunt restart-dataloss). + autoscaling_snapshot_hook: Option, + ec2_snapshot_hook: Option, } impl ContainerBackingHandles { @@ -561,9 +570,26 @@ impl ContainerBackingHandles { mq_runtime: p.mq_runtime.clone(), kafka_state: p.kafka_state.clone(), kafka_runtime: p.kafka_runtime.clone(), + autoscaling_snapshot_hook: None, + ec2_snapshot_hook: None, } } + /// Attach the autoscaling + EC2 snapshot hooks so a detached ASG capacity + /// reconcile can persist the instances it launches (and their EC2 records) + /// once it completes, instead of relying on the racing + /// `persist_touched_services` that runs before the reconcile finishes. The + /// hooks come from `CloudFormationService::snapshot_hooks`; `None` (memory + /// mode) leaves the reconcile's persist a no-op. + pub(crate) fn with_snapshot_hooks( + mut self, + hooks: &BTreeMap<&'static str, SnapshotHook>, + ) -> Self { + self.autoscaling_snapshot_hook = hooks.get("autoscaling").cloned(); + self.ec2_snapshot_hook = hooks.get("ec2").cloned(); + self + } + /// Back each freshly-inserted container resource with a REAL container in a /// detached task, so the stack op never blocks on a container boot/pull (the /// #1539/#1730 timeout lesson). Drains the provisioner's queued spawn intents. @@ -593,6 +619,10 @@ impl ContainerBackingHandles { let ec2_runtime = self.ec2_runtime.clone(); let account = self.account_id.clone(); let region = self.region.clone(); + let persist = fakecloud_autoscaling::cfn_provision::CfnReconcilePersistHooks { + autoscaling: self.autoscaling_snapshot_hook.clone(), + ec2: self.ec2_snapshot_hook.clone(), + }; tokio::spawn(async move { fakecloud_autoscaling::cfn_provision::cfn_reconcile_capacity( asg_state, @@ -601,6 +631,7 @@ impl ContainerBackingHandles { group_name, account, region, + persist, ) .await; }); @@ -1024,7 +1055,8 @@ impl CloudFormationService { // Reject a TypeName fakecloud has no provisioner for, so Cloud Control // never records a resource with no backing service state. provisioner.strict_unknown_types = true; - let backing = ContainerBackingHandles::from_provisioner(&provisioner); + let backing = ContainerBackingHandles::from_provisioner(&provisioner) + .with_snapshot_hooks(&self.snapshot_hooks); let spawns = provisioner.pending_container_spawns.clone(); let def = template::ResourceDefinition { logical_id: "Resource".to_string(), @@ -1671,7 +1703,8 @@ impl CloudFormationService { // into spawn_blocking, so the post-provision drain (below) can back // freshly-inserted container resources with REAL containers. let container_spawns = provisioner.pending_container_spawns.clone(); - let backing_handles = ContainerBackingHandles::from_provisioner(&provisioner); + let backing_handles = ContainerBackingHandles::from_provisioner(&provisioner) + .with_snapshot_hooks(&snapshot_hooks); // The provisioning loop is fully synchronous (it may block on cold // image pulls / custom-resource Lambda invokes). Hand it to a @@ -2355,7 +2388,8 @@ impl CloudFormationService { // reaching the update path), then run any deferred custom-resource // invokes. { - let handles = ContainerBackingHandles::from_provisioner(&provisioner); + let handles = ContainerBackingHandles::from_provisioner(&provisioner) + .with_snapshot_hooks(&self.snapshot_hooks); handles.spawn_container_intents(std::mem::take( &mut *provisioner.pending_container_spawns.lock(), )); diff --git a/crates/fakecloud-ec2/src/service/mod.rs b/crates/fakecloud-ec2/src/service/mod.rs index 7b4831cd3..54c201f8f 100644 --- a/crates/fakecloud-ec2/src/service/mod.rs +++ b/crates/fakecloud-ec2/src/service/mod.rs @@ -1057,6 +1057,8 @@ impl Ec2Service { } } + // (helper `recovery_terminal_wins` is defined at module scope below.) + pub async fn recover_persisted_containers(&self) { let Some(runtime) = self.runtime.clone() else { // Metadata-only mode: there is no container to reconcile, but an @@ -1161,8 +1163,16 @@ impl Ec2Service { running, ) { (Some(inst), Ok(r)) => { - if inst.state_code == 48 { - // Terminated during recovery: drop the container. + // A concurrent Terminate (code 48) or Stop (code 80) + // during the recovery window wins: don't resurrect the + // instance to `running`, and drop the container we just + // started (a terminated/stopped instance owns none). + // Mirrors the both-terminal guard in + // `instance::reconcile_started`. Guarding only code 48 + // let a concurrent StopInstances get clobbered back to + // `running` with a fresh container (bug-hunt + // restart-dataloss). + if recovery_terminal_wins(inst.state_code) { true } else { inst.state_code = 16; @@ -1173,15 +1183,25 @@ impl Ec2Service { } } (Some(inst), Err(error)) => { - tracing::error!( - %error, - instance_id = %p.id, - "failed to recover ec2 backing container after restart", - ); - inst.state_code = 80; - inst.state_name = "stopped".to_string(); - inst.container_id = None; - false + // A concurrent Terminate (code 48) or Stop (code 80) + // wins over a boot failure too: leave the terminal + // state as-is rather than overwriting it to `stopped`. + // The unguarded write here previously clobbered a + // concurrent TerminateInstances (code 48) to `stopped` + // (bug-hunt restart-dataloss). + if recovery_terminal_wins(inst.state_code) { + false + } else { + tracing::error!( + %error, + instance_id = %p.id, + "failed to recover ec2 backing container after restart", + ); + inst.state_code = 80; + inst.state_name = "stopped".to_string(); + inst.container_id = None; + false + } } // Deleted during recovery: stop the container we just // started so it isn't orphaned (mirrors ElastiCache). @@ -1255,6 +1275,18 @@ pub async fn save_ec2_snapshot( } } +/// True if a concurrent Terminate (code 48) or Stop (code 80) has already +/// claimed an instance during the restart recovery window, so +/// `recover_persisted_containers` must NOT overwrite that terminal state with a +/// recovered `running`/`stopped` row. Mirrors the both-terminal guard in +/// [`instance::reconcile_started`]; applies to BOTH the boot-success and +/// boot-failure arms (a boot failure previously clobbered a concurrent +/// Terminate to `stopped`; a boot success clobbered a concurrent Stop back to +/// `running` with a fresh container). bug-hunt restart-dataloss. +fn recovery_terminal_wins(state_code: i64) -> bool { + state_code == 48 || state_code == 80 +} + #[async_trait] impl AwsService for Ec2Service { fn service_name(&self) -> &str { @@ -2705,3 +2737,104 @@ impl Ec2Service { ) } } + +#[cfg(test)] +mod recover_guard_tests { + use super::recovery_terminal_wins; + + // A minimal stand-in for the `state_code`/`state_name` fields the recovery + // closure mutates. Kept independent of the full `Instance` struct so this + // regression stays pinned to the guard behavior, not to unrelated field + // churn (the struct-field repo-wide-ctor hazard). + struct Row { + state_code: i64, + state_name: &'static str, + } + + // Boot-success arm of `recover_persisted_containers`, driven by the guard. + // Returns whether the just-started container must be reaped. + fn apply_boot_success(row: &mut Row) -> bool { + if recovery_terminal_wins(row.state_code) { + true + } else { + row.state_code = 16; + row.state_name = "running"; + false + } + } + + // Boot-failure arm of `recover_persisted_containers`, driven by the guard. + fn apply_boot_failure(row: &mut Row) -> bool { + if recovery_terminal_wins(row.state_code) { + false + } else { + row.state_code = 80; + row.state_name = "stopped"; + false + } + } + + // Instance state codes: 0 pending, 16 running, 48 terminated, 80 stopped. + #[test] + fn terminal_states_win_over_recovery() { + assert!(recovery_terminal_wins(48), "terminated wins"); + assert!(recovery_terminal_wins(80), "stopped wins"); + assert!(!recovery_terminal_wins(16), "running is recoverable"); + assert!(!recovery_terminal_wins(0), "pending is recoverable"); + } + + #[test] + fn boot_success_does_not_resurrect_concurrent_stop() { + // Concurrent StopInstances set code 80 during the recovery window. + let mut row = Row { + state_code: 80, + state_name: "stopped", + }; + let reap = apply_boot_success(&mut row); + assert!(reap, "the just-started container must be reaped"); + assert_eq!(row.state_code, 80, "stop must not be clobbered to running"); + assert_eq!(row.state_name, "stopped"); + } + + #[test] + fn boot_failure_does_not_clobber_concurrent_terminate() { + // Concurrent TerminateInstances set code 48 during the recovery window, + // and the backing container failed to boot. + let mut row = Row { + state_code: 48, + state_name: "terminated", + }; + let reap = apply_boot_failure(&mut row); + assert!(!reap, "boot failed: nothing to reap"); + assert_eq!( + row.state_code, 48, + "terminate must not be overwritten to stopped" + ); + assert_eq!(row.state_name, "terminated"); + } + + #[test] + fn boot_success_recovers_a_still_pending_instance() { + // No concurrent op: a persisted pending row recovers to running. + let mut row = Row { + state_code: 0, + state_name: "pending", + }; + let reap = apply_boot_success(&mut row); + assert!(!reap); + assert_eq!(row.state_code, 16, "pending recovers to running"); + assert_eq!(row.state_name, "running"); + } + + #[test] + fn boot_failure_stops_a_still_pending_instance() { + // No concurrent op, boot failed: a pending row is marked stopped. + let mut row = Row { + state_code: 0, + state_name: "pending", + }; + let reap = apply_boot_failure(&mut row); + assert!(!reap); + assert_eq!(row.state_code, 80, "failed boot marks stopped"); + } +} diff --git a/crates/fakecloud-kinesis/src/delivery.rs b/crates/fakecloud-kinesis/src/delivery.rs index b526569ad..74a85a06d 100644 --- a/crates/fakecloud-kinesis/src/delivery.rs +++ b/crates/fakecloud-kinesis/src/delivery.rs @@ -285,6 +285,41 @@ mod tests { ); } + // bug-hunt restart-dataloss: the EventBridge Scheduler and EventBridge-Pipes + // delivery buses must build their Kinesis sender with `with_dirty_flag`, not + // `::new`. `::new` leaves `dirty: None`, so `mark_dirty()` is a no-op and a + // record delivered to a Kinesis stream target by a schedule/pipe lands in + // memory but never flips the shared flag the flusher persists on -> the + // record vanishes on restart. This locks in the `::new` vs `with_dirty_flag` + // contract those wiring sites depend on. + #[test] + fn new_does_not_mark_dirty_but_with_dirty_flag_does() { + let stream = make_stream("contract", 1); + let arn = stream.stream_arn.clone(); + let encoded = base64::engine::general_purpose::STANDARD.encode(b"evt"); + + // `::new` (the buggy scheduler/pipes wiring) delivers but never signals + // the flusher, so the record would be lost on restart. + let state_new = make_state(make_stream("contract", 1)); + let flag = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let via_new = KinesisDeliveryImpl::new(state_new.clone()); + via_new.put_record(&arn, &encoded, "pk"); + assert!( + !flag.load(Ordering::Acquire), + "::new must not touch any external dirty flag" + ); + + // `with_dirty_flag` (the fixed wiring) flips the shared flag so the + // background flusher persists the delivery. + let state_dirty = make_state(stream); + let via_flag = KinesisDeliveryImpl::with_dirty_flag(state_dirty.clone(), flag.clone()); + via_flag.put_record(&arn, &encoded, "pk"); + assert!( + flag.load(Ordering::Acquire), + "with_dirty_flag must flip the shared flag so the flusher persists" + ); + } + // bug-audit 2026-06-26, 4.1: after a split/merge the modulo router could // land a fanned-in record on a CLOSED parent shard (drained, null iterator) // -> silent loss. Delivery must route by hash range over OPEN shards only. diff --git a/crates/fakecloud-server/src/main.rs b/crates/fakecloud-server/src/main.rs index 2bd411280..a72272bba 100644 --- a/crates/fakecloud-server/src/main.rs +++ b/crates/fakecloud-server/src/main.rs @@ -4216,7 +4216,13 @@ async fn main() { }; let mut autoscaling_service = fakecloud_autoscaling::AutoScalingService::new(autoscaling_state.clone()) - .with_ec2(ec2_state.clone(), ec2_runtime.clone()); + .with_ec2(ec2_state.clone(), ec2_runtime.clone()) + // ASG capacity reconciliation launches REAL EC2 instances through a + // bare Ec2Service; without the EC2 snapshot hook those records live + // only in memory and leak their containers on restart (the EC2 + // boot-recovery has no persisted row to re-drive). The "ec2" hook was + // registered above when the EC2 service was wired. + .with_ec2_snapshot_hook(cfn_snapshot_hooks.get("ec2").cloned()); if let Some(store) = autoscaling_snapshot_store.clone() { autoscaling_service = autoscaling_service.with_snapshot_store(store); } @@ -6613,8 +6619,15 @@ async fn main() { ) }; let delivery_for_scheduler = { + // Scheduler-driven Kinesis deliveries must flip the shared dirty flag so + // the background delivery flusher persists them; `::new` leaves the flag + // `None` and records delivered to a Kinesis stream target by a schedule + // would vanish on restart (bug-hunt restart-dataloss). let kinesis_delivery_for_scheduler = - fakecloud_kinesis::delivery::KinesisDeliveryImpl::new(kinesis_state.clone()); + fakecloud_kinesis::delivery::KinesisDeliveryImpl::with_dirty_flag( + kinesis_state.clone(), + kinesis_delivery_dirty.clone(), + ); let ses_dispatcher_for_scheduler: Arc< dyn fakecloud_core::delivery::SesSendEmailDispatcher, > = Arc::new(SesSendEmailDispatcherImpl { @@ -6725,8 +6738,15 @@ async fn main() { .with_registry(sfn_registry_handle.clone()), ) }; + // Pipe-driven Kinesis deliveries must flip the shared dirty flag so the + // background delivery flusher persists them; `::new` leaves the flag + // `None` and records delivered to a Kinesis stream target by a pipe would + // vanish on restart (bug-hunt restart-dataloss). let pipes_kinesis_delivery = - fakecloud_kinesis::delivery::KinesisDeliveryImpl::new(kinesis_state.clone()); + fakecloud_kinesis::delivery::KinesisDeliveryImpl::with_dirty_flag( + kinesis_state.clone(), + kinesis_delivery_dirty.clone(), + ); let mut pipes_bus = DeliveryBus::new() .with_sqs(sqs_delivery.clone()) .with_sns(sns_delivery.clone())