fix(ecr,sqs,elasticache): data-durability + portability gaps (bug-hunt 4.2/4.3/4.4) (#2290)

Author: vieiralucas

crates/fakecloud-ecr/src/oci.rs

@@ -52,7 +52,12 @@ pub(crate) fn create_upload_spool(upload_id: &str) -> std::io::Result<PathBuf> {
 /// `UploadLayerPart` route, where the SDK sends a base64-encoded chunk.
 pub(crate) fn append_bytes_sync(path: &Path, bytes: &[u8]) -> std::io::Result<()> {
     use std::io::Write;
-    let mut f = std::fs::OpenOptions::new().append(true).open(path)?;
+    // `create(true)`: a spool lost to a restart (temp-dir, non-durable) is
+    // recreated rather than 500-ing the chunk.
+    let mut f = std::fs::OpenOptions::new()
+        .append(true)
+        .create(true)
+        .open(path)?;
     f.write_all(bytes)?;
     f.sync_data()?;
     Ok(())
@@ -69,6 +74,7 @@ pub(crate) async fn append_stream(
     use tokio::io::AsyncWriteExt;
     let mut file = tokio::fs::OpenOptions::new()
         .append(true)
+        .create(true)
         .open(path)
         .await
         .map_err(|e| {

crates/fakecloud-ecr/src/state.rs

@@ -45,7 +45,14 @@ pub struct EcrState {
     pub account_settings: BTreeMap<String, String>,
     /// Layer upload state machine keyed by `uploadId`. Each entry is
     /// tied to a specific repository.
-    #[serde(default)]
+    ///
+    /// NOT persisted: the per-upload spool file lives under the system temp
+    /// dir (`spool_path`) and never survives a restart, so persisting the
+    /// upload metadata alone would leave orphan entries whose next
+    /// `UploadLayerPart` opens a missing spool and 500s. Dropping in-flight
+    /// uploads on restart keeps the map consistent with the (gone) spools; a
+    /// stale client simply re-initiates, matching ECR's finite upload lifetime.
+    #[serde(skip)]
     pub layer_uploads: BTreeMap<String, LayerUpload>,
     /// Pull-time update exclusions keyed by IAM principal ARN. These
     /// are registry-level per the Smithy model.
@@ -411,3 +418,39 @@ pub struct ReplicationDestination {
     pub region: String,
     pub registry_id: String,
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn layer_uploads_are_not_persisted_across_snapshot() {
+        // In-progress uploads spool to the system temp dir, which does not
+        // survive a restart. Persisting the upload metadata alone would leave
+        // an orphan entry whose next `UploadLayerPart` opens a missing spool
+        // and 500s (bug-hunt 4.2), so `layer_uploads` is `#[serde(skip)]`.
+        let mut state = EcrState::new("123456789012", "us-east-1");
+        state.layer_uploads.insert(
+            "upload-1".to_string(),
+            LayerUpload {
+                upload_id: "upload-1".to_string(),
+                repository_name: "repo".to_string(),
+                created_at: Utc::now(),
+                spool_path: "/tmp/fakecloud-ecr-upload-upload-1".to_string(),
+                last_byte_received: 42,
+            },
+        );
+
+        let json = serde_json::to_string(&state).unwrap();
+        assert!(
+            !json.contains("upload-1"),
+            "in-progress uploads must not be serialized"
+        );
+
+        let restored: EcrState = serde_json::from_str(&json).unwrap();
+        assert!(
+            restored.layer_uploads.is_empty(),
+            "in-progress uploads must not survive a restart"
+        );
+    }
+}

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

@@ -3,6 +3,21 @@
 
 use super::*;
 
+/// Per-snapshot RDB dump path. Roots under the platform temp dir rather than a
+/// hardcoded `/tmp` (absent on Windows, sometimes non-writable in containers).
+/// The PID keeps concurrent servers from colliding on the same file.
+fn snapshot_rdb_temp_path(account_id: &str, snapshot_name: &str) -> String {
+    std::env::temp_dir()
+        .join(format!(
+            "fakecloud-ec-{}-{}-{}.rdb",
+            account_id,
+            snapshot_name,
+            std::process::id()
+        ))
+        .to_string_lossy()
+        .into_owned()
+}
+
 impl ElastiCacheService {
     pub(super) fn create_serverless_cache_snapshot(
         &self,
@@ -289,12 +304,7 @@ impl ElastiCacheService {
         };
 
         if let Some(ref runtime) = self.runtime {
-            let tmp_path = format!(
-                "/tmp/fakecloud-ec-{}-{}-{}.rdb",
-                request.account_id,
-                snapshot_name,
-                std::process::id()
-            );
+            let tmp_path = snapshot_rdb_temp_path(&request.account_id, &snapshot_name);
             match runtime.dump_rdb(&group_id, &tmp_path).await {
                 Ok(()) => {
                     snapshot.rdb_path = Some(tmp_path);
@@ -556,3 +566,23 @@ impl ElastiCacheService {
         ))
     }
 }
+
+#[cfg(test)]
+mod snapshot_path_tests {
+    use super::snapshot_rdb_temp_path;
+
+    #[test]
+    fn rdb_temp_path_roots_under_platform_temp_dir() {
+        // Portability (4.4): the RDB dump must live under the platform temp
+        // dir, not a hardcoded `/tmp` that is absent on Windows / some
+        // containers.
+        let path = snapshot_rdb_temp_path("123456789012", "nightly");
+        let temp = std::env::temp_dir();
+        assert!(
+            std::path::Path::new(&path).starts_with(&temp),
+            "{path} must root under {}",
+            temp.display()
+        );
+        assert!(path.contains("nightly") && path.ends_with(".rdb"));
+    }
+}

crates/fakecloud-sqs/src/service/messages.rs

@@ -812,6 +812,17 @@ impl SqsService {
                     "redrive target queue not found; returning message to source queue"
                 );
                 if let Some(source) = state.queues.values_mut().find(|q| q.arn == queue_arn) {
+                    // Reset the receive history before returning the message.
+                    // It re-enters the source with `receive_count` still past
+                    // `maxReceiveCount` (that is what triggered the redrive) and
+                    // `visible_at = None` (immediately visible), so without this
+                    // the very next ReceiveMessage would re-redrive it, hit the
+                    // same unresolvable target, and spin in a hot loop. Resetting
+                    // gives the consumer a fresh `maxReceiveCount` attempts —
+                    // the message is reprocessed rather than tight-looping, and
+                    // is still never destroyed.
+                    msg.receive_count = 0;
+                    msg.first_received_at = None;
                     source.messages.push_back(msg);
                 }
             }

crates/fakecloud-sqs/src/service_tests.rs

@@ -1394,6 +1394,18 @@ fn redrive_with_unresolvable_dlq_arn_keeps_message_on_source_queue() {
         .unwrap(),
     );
     assert_eq!(attrs["Attributes"]["ApproximateNumberOfMessages"], "1");
+
+    // Loop-prevention (4.3): the message was returned to source with its
+    // receive history reset, so a fresh receive delivers it again for normal
+    // reprocessing. Before the fix its receive_count stayed past
+    // maxReceiveCount with visible_at=None, so the next receive re-redrove it
+    // into the same unresolvable target and spun in a hot loop (0 delivered
+    // forever).
+    assert_eq!(
+        receive_msgs(&svc, &src_url, 1).len(),
+        1,
+        "returned message must be re-receivable, not hot-looping through redrive"
+    );
 }
 
 // ── Error branch tests ──