diff --git a/src/engine/tests.rs b/src/engine/tests.rs index 39e279f4..2990b539 100644 --- a/src/engine/tests.rs +++ b/src/engine/tests.rs @@ -662,6 +662,11 @@ fn should_preserve_partitioned_compaction_across_simulated_cloud_reopen() -> Mid OpenOptions::cloud_simulated(temp_dir.path(), "partitioned-bucket", "partitioned-prefix") .background_compaction(false) .with_memtable_size_limit(64 * 1024) + // This is a multi-output durability qualification, not a response + // deadline test. Windows hosted runners can execute it alongside + // the intentionally large compaction resource proof, so retain a + // bounded but explicit allowance for that filesystem contention. + .runtime_response_timeout(Duration::from_mins(3)) .build() }); diff --git a/src/runtime/hybrid_persistence.rs b/src/runtime/hybrid_persistence.rs index 3aeab98c..cd9c6298 100644 --- a/src/runtime/hybrid_persistence.rs +++ b/src/runtime/hybrid_persistence.rs @@ -860,6 +860,9 @@ impl HybridPersistence for HybridStorage { } fn delete_sst_object_blocking(&self, sst_name: &str) -> MidgeResult<()> { + crate::failpoints::fail_point!("midge::cloud::inject_fail_sst_delete", |_| Err( + MidgeError::Internal("failpoint: cloud SST delete failed".to_string()) + )); self.delete_immutable_object_blocking(&crate::sst::object_key(sst_name)) } } diff --git a/tests/backpressure.rs b/tests/backpressure.rs index 65d6fb84..c783d533 100644 --- a/tests/backpressure.rs +++ b/tests/backpressure.rs @@ -6,8 +6,8 @@ use cntryl_midge::{ }; use common::*; use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::{Arc, Barrier, Mutex, OnceLock}; -use std::time::Duration; +use std::sync::{Arc, Mutex, OnceLock}; +use std::time::{Duration, Instant}; use tempfile::TempDir; static BACKPRESSURE_STRESS_TEST_LOCK: OnceLock> = OnceLock::new(); @@ -334,7 +334,7 @@ fn should_prevent_oom_by_rejecting_writes_when_budget_exceeded() { .unwrap_or_else(std::sync::PoisonError::into_inner); // Act - let results = std::cell::RefCell::new(Vec::<(String, u64, usize, bool)>::new()); + let results = std::cell::RefCell::new(Vec::<(String, u64, u64, usize, bool)>::new()); for_each_storage_mode(&["local"], |mode, opts| { let mut opts = opts; opts = opts.memory_budget(512 * 1024); // 512KB instead of 2MB for faster backpressure trigger @@ -350,60 +350,49 @@ fn should_prevent_oom_by_rejecting_writes_when_budget_exceeded() { let cf = engine.create_column_family("test").expect("create cf"); let cf_id = cf.id(); - let worker_count = 1; - let max_attempts_per_worker = 64; - let shutdown = Arc::new(AtomicBool::new(false)); - let barrier = Arc::new(Barrier::new(worker_count)); - let mut handles = Vec::new(); - - for worker_id in 0..worker_count { - let shutdown_clone = shutdown.clone(); - let engine_clone = Arc::clone(&engine); - let barrier_clone = Arc::clone(&barrier); - let write_options = buffered_write_options(mode); + let max_attempts = 64; + let write_options = buffered_write_options(mode); + let mut total_writes = 0; + let mut total_stalls = 0; - handles.push(std::thread::spawn(move || { - let mut total_writes = 0; - let mut total_stalls = 0; - - barrier_clone.wait(); - while !shutdown_clone.load(Ordering::Relaxed) { - let key = format!("worker_{worker_id}_key_{total_writes}"); - let value = vec![0u8; 8192]; // 8KB for faster memory budget exhaustion - let mut txn = engine_clone - .begin_tx(cf_id, TransactionMode::ReadWrite) - .expect("begin"); - txn.put(key.as_bytes().to_vec(), value.clone(), None) - .expect("put"); - - match txn.commit(write_options) { - Ok(()) => total_writes += 1, - Err(MidgeError::WriteStall(_)) => { - total_stalls += 1; - std::thread::sleep(Duration::from_millis(10)); - } - Err(e) => panic!("unexpected: {e:?}"), - } + while total_writes + total_stalls < max_attempts { + let key = format!("key_{total_writes}"); + let value = vec![0u8; 8192]; // 8KB for faster memory budget exhaustion + let mut txn = engine + .begin_tx(cf_id, TransactionMode::ReadWrite) + .expect("begin"); + txn.put(key.as_bytes().to_vec(), value, None).expect("put"); - if total_writes + total_stalls >= max_attempts_per_worker { - break; - } + match txn.commit(write_options) { + Ok(()) => total_writes += 1, + Err(MidgeError::WriteStall(_)) => { + total_stalls += 1; + std::thread::sleep(Duration::from_millis(10)); } - - (total_writes, total_stalls) - })); + Err(e) => panic!("unexpected: {e:?}"), + } } - std::thread::sleep(Duration::from_secs(2)); - shutdown.store(true, Ordering::Relaxed); - let total_stalls = handles - .into_iter() - .map(|handle| handle.join().expect("panic").1) - .sum(); + // Natural flush publication is asynchronous. Wait for the actual acceptance + // condition instead of assuming a fixed sleep covers hosted-runner contention. + let deadline = Instant::now() + Duration::from_secs(30); + let metrics = loop { + let metrics = engine.get_runtime_metrics().expect("runtime metrics"); + if total_stalls > 0 || (metrics.sst_count > 0 && !metrics.write_stalled) { + break metrics; + } + assert!( + Instant::now() < deadline, + "local pressure was not rejected or naturally flushed: writes={total_writes}, stalls={total_stalls}, ssts={}, write_stalled={}", + metrics.sst_count, + metrics.write_stalled + ); + std::thread::sleep(Duration::from_millis(10)); + }; - let metrics = engine.get_runtime_metrics().expect("runtime metrics"); results.borrow_mut().push(( mode.to_string(), + total_writes, total_stalls, metrics.sst_count, metrics.write_stalled, @@ -411,11 +400,11 @@ fn should_prevent_oom_by_rejecting_writes_when_budget_exceeded() { }); // Assert - for (mode, total_stalls, sst_count, write_stalled) in results.into_inner() { + for (mode, total_writes, total_stalls, sst_count, write_stalled) in results.into_inner() { assert_eq!(mode, "local"); assert!( total_stalls > 0 || (sst_count > 0 && !write_stalled), - "Expected local mode to either reject writes under hard pressure or relieve pressure via natural flush" + "Expected local mode to either reject writes under hard pressure or relieve pressure via natural flush: writes={total_writes}, stalls={total_stalls}, ssts={sst_count}, write_stalled={write_stalled}" ); } } diff --git a/tests/engine_gc_cloud.rs b/tests/engine_gc_cloud.rs index a0478f80..9eaf6dc7 100644 --- a/tests/engine_gc_cloud.rs +++ b/tests/engine_gc_cloud.rs @@ -189,20 +189,13 @@ fn should_not_collect_cloud_objects_referenced_by_manifest() { } /// Simulates a cloud provider outage that specifically affects deleting -/// GC'd (orphaned) objects, without disturbing anything else, by chmod-ing -/// the simulated cloud bucket read-only for the exact window between -/// compaction's manifest publish and its orphan cleanup. This mirrors -/// `should_handle_cloud_unavailable_during_eviction` in -/// `tests/hybrid_storage.rs`, which uses the same "the simulated cloud -/// backend is just a filesystem directory" trick for upload outages; here -/// we pin the outage to the GC boundary using the existing -/// `slice6::after_manifest_persist_before_sst_gc` failpoint so the prior -/// (successful) upload of the compacted output isn't itself blocked. +/// GC'd (orphaned) objects, without disturbing the output upload that must +/// precede manifest publication. A provider-boundary failpoint keeps this +/// deterministic even when the process has permission to delete read-only +/// files, as root does inside the Docker qualification image. #[test] -#[cfg(all(feature = "failpoints", unix))] +#[cfg(feature = "failpoints")] fn should_handle_gc_when_cloud_delete_fails() { - use std::os::unix::fs::PermissionsExt; - let _guard = failpoint_test_lock() .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); @@ -217,7 +210,7 @@ fn should_handle_gc_when_cloud_delete_fails() { .build() .expect("build simulated cloud options"); let l0_batch_size = options.l0_compaction_trigger(); - let engine = Engine::open(options).expect("open simulated cloud engine"); + let mut engine = Engine::open(options).expect("open simulated cloud engine"); let cf = engine.create_column_family("test").expect("create cf"); // Write exactly one configured L0 batch. This isolates the delete failure @@ -247,34 +240,16 @@ fn should_handle_gc_when_cloud_delete_fails() { "expected one configured L0 batch to be mirrored to cloud storage, got {before_objects:?}" ); - // Arm a failpoint at the exact boundary between compaction's manifest - // publish (which already uploaded the compacted output) and its - // orphan-SST GC pass, and make the bucket read-only there so only the - // orphan delete fails. + // Arm only the remote SST delete boundary. The compacted output upload + // and manifest authority switch remain real simulated-cloud operations. let scenario = fail::FailScenario::setup(); - let original_permissions = std::fs::metadata(&cloud_sst_dir) - .expect("stat cloud sst dir") - .permissions(); - let outage_dir = cloud_sst_dir.clone(); - fail::cfg_callback("slice6::after_manifest_persist_before_sst_gc", move || { - std::fs::set_permissions(&outage_dir, std::fs::Permissions::from_mode(0o500)) - .expect("simulate cloud delete outage via read-only bucket"); - }) - .expect("configure cloud delete outage failpoint"); + fail::cfg("midge::cloud::inject_fail_sst_delete", "return") + .expect("configure cloud delete outage failpoint"); // Act: compaction should orphan the selected input SSTs and try (and fail) - // to delete them from the now-read-only cloud bucket. + // to delete them from cloud storage. let compact_result = engine.compact_all(); - // Give the async cloud-delete worker time to attempt (and fail) the - // delete before we inspect the bucket. - thread::sleep(Duration::from_millis(300)); - - fail::remove("slice6::after_manifest_persist_before_sst_gc"); - std::fs::set_permissions(&cloud_sst_dir, original_permissions) - .expect("restore cloud sst permissions"); - scenario.teardown(); - // Assert: compaction tolerates the delete failure rather than // propagating it as an error. assert!( @@ -282,17 +257,6 @@ fn should_handle_gc_when_cloud_delete_fails() { "compact_all should tolerate a cloud delete failure: {compact_result:?}" ); - // Assert: the orphaned objects are still present in cloud storage - // because their delete genuinely failed and was retained for retry, - // not silently skipped or corrupted. - let after_objects = sst_object_names(&cloud_sst_dir); - let retained: Vec<_> = before_objects.intersection(&after_objects).collect(); - assert!( - !retained.is_empty(), - "expected the orphaned cloud objects whose delete failed to remain \ - in cloud storage for retry, got {after_objects:?}" - ); - // Assert: engine remains fully functional; no data was lost. let tx = engine .begin_tx(cf.id(), TransactionMode::ReadOnly) @@ -304,6 +268,27 @@ fn should_handle_gc_when_cloud_delete_fails() { "data lost after cloud delete failure" ); } + drop(tx); + + // Shutdown joins every cloud-delete worker while the outage remains + // armed, so the filesystem observation cannot race an unattempted delete. + engine + .shutdown(Duration::from_secs(10)) + .expect("shutdown after failed cloud delete"); + let after_objects = sst_object_names(&cloud_sst_dir); + let retained: Vec<_> = before_objects.intersection(&after_objects).collect(); + + fail::remove("midge::cloud::inject_fail_sst_delete"); + scenario.teardown(); + + // Assert: the orphaned objects are still present in cloud storage + // because their delete genuinely failed and was retained for retry, + // not silently skipped or corrupted. + assert!( + !retained.is_empty(), + "expected the orphaned cloud objects whose delete failed to remain \ + in cloud storage for retry, got {after_objects:?}" + ); eprintln!("✓ Engine gracefully handled cloud delete failure"); } diff --git a/tests/hybrid_storage.rs b/tests/hybrid_storage.rs index 70c15908..4b7ed7ff 100644 --- a/tests/hybrid_storage.rs +++ b/tests/hybrid_storage.rs @@ -16,7 +16,7 @@ //! should__given__when_ mod common; -#[cfg(unix)] +#[cfg(feature = "failpoints")] use cntryl_midge::EngineHealth; use cntryl_midge::{Engine, MidgeError, OpenOptions, TransactionMode, WriteOptions}; use common::*; @@ -24,6 +24,9 @@ use std::sync::Arc; use std::thread; use std::time::Duration; +#[cfg(feature = "failpoints")] +const CLOUD_OUTAGE_CHILD_ENV: &str = "MIDGE_HYBRID_CLOUD_OUTAGE_CHILD"; + /// Count files nested anywhere under `root`, used to prove that the /// filesystem-backed simulated cloud/local stores actually received or lost /// data, rather than trusting a `get()` result alone. @@ -56,25 +59,6 @@ fn incompressible_value(len: usize, seed: u8) -> Vec { .collect() } -/// Recursively apply a permission mode to every directory under (and -/// including) `root`. Unix directory-write checks are per-directory, so -/// blocking writes anywhere under a simulated cloud bucket (not just its -/// top-level directory) requires chmod'ing every subdirectory that already -/// exists, such as the WAL directory created at engine-open time. -#[cfg(unix)] -fn set_dir_permissions_recursive(root: &std::path::Path, mode: u32) { - use std::os::unix::fs::PermissionsExt; - if let Ok(entries) = std::fs::read_dir(root) { - for entry in entries.flatten() { - let path = entry.path(); - if path.is_dir() { - set_dir_permissions_recursive(&path, mode); - } - } - } - let _ = std::fs::set_permissions(root, std::fs::Permissions::from_mode(mode)); -} - // ============================================================================ // TEST GROUP: Memory Budget & Eviction Control // ============================================================================ @@ -509,15 +493,34 @@ fn should_persist_eviction_state_across_restart() { } #[test] -#[cfg(unix)] +#[cfg(feature = "failpoints")] fn should_handle_cloud_unavailable_during_eviction() { - // "local" storage mode has no cloud tier and no upload to fail, so the - // scenario this test names was never exercised. The simulated cloud - // backend is itself just a filesystem directory (see - // `storage::test_support::build_cloud_backed_filesystem_simulation`), so - // a real outage can be reproduced by making that directory unwritable - // right before the flush/eviction pipeline tries to upload to it - - // without inventing any new production fault-injection API. + // Failpoints are process-global. Run the injection in an exact-test child + // so parallel tests in this binary cannot observe the simulated outage. + if std::env::var_os(CLOUD_OUTAGE_CHILD_ENV).is_none() { + let output = std::process::Command::new( + std::env::current_exe().expect("locate hybrid storage test executable"), + ) + .arg("--exact") + .arg("should_handle_cloud_unavailable_during_eviction") + .arg("--nocapture") + .arg("--test-threads=1") + .env(CLOUD_OUTAGE_CHILD_ENV, "1") + .output() + .expect("run isolated cloud outage child"); + assert!( + output.status.success(), + "isolated cloud outage child failed: status={} stdout={} stderr={}", + output.status, + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + return; + } + + // "local" storage mode has no cloud tier and no upload to fail. Exercise + // the simulated-cloud provider boundary directly so the outage remains + // deterministic even when this test runs as root in the Docker image. let temp_dir = test_temp_dir(); let budget_bytes = 256 * 1024; // small budget forces eviction attempts let opts = OpenOptions::cloud_simulated(temp_dir.path(), "test-bucket", "test-prefix") @@ -539,35 +542,31 @@ fn should_handle_cloud_unavailable_during_eviction() { } tx.commit(WriteOptions::cloud_async()).expect("commit"); - // Simulate a cloud outage: make every directory under the simulated - // cloud bucket unwritable before eviction tries to upload into it. This - // has to be recursive (not just the top-level bucket dir) because the - // WAL subdirectory already exists with its own write permission. - let cloud_store_dir = temp_dir.path().join("cloud_store"); - std::fs::create_dir_all(&cloud_store_dir).expect("cloud store dir exists"); - let files_before_outage_attempt = count_files_recursive(&cloud_store_dir); - set_dir_permissions_recursive(&cloud_store_dir, 0o500); + let cloud_sst_dir = temp_dir.path().join("cloud_store").join("sst"); + let files_before_outage_attempt = count_files_recursive(&cloud_sst_dir); + let scenario = fail::FailScenario::setup(); + fail::cfg("midge::cloud::inject_fail_sst_upload", "return") + .expect("configure cloud SST upload outage"); - // Act: Flush with cloud "down" (uploads should fail, not panic) - let flush_result = engine.flush_cf(&cf); - thread::sleep(Duration::from_millis(200)); + // Act: Flush with the remote SST provider unavailable. + let flush_error = engine + .flush_cf(&cf) + .expect_err("cloud outage should fail the SST upload"); + let files_after_outage_attempt = count_files_recursive(&cloud_sst_dir); - // Restore the bucket so later ops (and temp-dir cleanup) can proceed. - set_dir_permissions_recursive(&cloud_store_dir, 0o755); + fail::remove("midge::cloud::inject_fail_sst_upload"); + scenario.teardown(); - // Assert: the outage genuinely blocked the upload (no new object landed - // in the bucket while it was read-only) rather than trivially succeeding. + // Assert: the provider boundary genuinely rejected the upload and no new + // SST object landed in cloud storage. + assert!( + matches!(&flush_error, MidgeError::Internal(message) if message.contains("cloud SST upload failed")), + "unexpected cloud upload error: {flush_error:?}" + ); assert_eq!( - count_files_recursive(&cloud_store_dir), - files_before_outage_attempt, - "expected no objects to be written to the cloud store while it was unwritable" + files_after_outage_attempt, files_before_outage_attempt, + "expected no SST objects to be written while the provider was unavailable" ); - if let Err(err) = &flush_result { - assert!( - !matches!(err, MidgeError::Internal(msg) if msg.contains("panic")), - "flush surfaced an internal panic-shaped error during the outage: {err:?}" - ); - } // Assert: Engine still operational and data stayed available locally. let tx = engine @@ -598,7 +597,7 @@ fn should_handle_cloud_unavailable_during_eviction() { ); eprintln!( - "✓ Handled cloud unavailability gracefully; {accessible} keys still accessible, flush_result={flush_result:?}" + "✓ Handled cloud unavailability gracefully; {accessible} keys still accessible, flush_error={flush_error:?}" ); }