Skip to content

test: stabilize qualification closeout #277

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
smiggleworth merged 4 commits into from
Sep 4, 2026
Merged

test: stabilize qualification closeout #277

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
bc9138a
test: stabilize cloud qualification closeout
smiggleworth Sep 4, 2026
c047017
test: isolate cloud upload outage
smiggleworth Sep 4, 2026
498098d
test: wait for backpressure outcome
smiggleworth Sep 4, 2026
ecfd4ef
test: keep timeout MSRV-clean
smiggleworth Sep 4, 2026
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
5 changes: 5 additions & 0 deletions src/engine/tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()
});

Expand Down
3 changes: 3 additions & 0 deletions src/runtime/hybrid_persistence.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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))
}
}
Expand Down
91 changes: 40 additions & 51 deletions tests/backpressure.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Expand Down Expand Up @@ -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
Expand All @@ -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}"
);
}
}
Expand Down
79 changes: 32 additions & 47 deletions tests/engine_gc_cloud.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand All @@ -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
Expand Down Expand Up @@ -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)
Expand All @@ -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");
}
Expand Down
Loading