test: stabilize qualification closeout (#277)

* test: stabilize cloud qualification closeout

* test: isolate cloud upload outage

* test: wait for backpressure outcome

* test: keep timeout MSRV-clean

Author: smiggleworth

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()
     });
 

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))
     }
 }

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<Mutex<()>> = 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,72 +350,61 @@ 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,
         ));
     });
 
     // 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}"
         );
     }
 }

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,52 +240,23 @@ 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!(
         compact_result.is_ok(),
         "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");
 }