Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 7 additions & 1 deletion crates/fakecloud-ecr/src/oci.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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(())
Expand All @@ -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| {
Expand Down
45 changes: 44 additions & 1 deletion crates/fakecloud-ecr/src/state.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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"
);
}
}
42 changes: 36 additions & 6 deletions crates/fakecloud-elasticache/src/service/snapshots.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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"));
}
}
11 changes: 11 additions & 0 deletions crates/fakecloud-sqs/src/service/messages.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
}
Expand Down
12 changes: 12 additions & 0 deletions crates/fakecloud-sqs/src/service_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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 ──
Expand Down
Loading