Files
foxhunt/crates/ml/tests/recovery_tests.rs
jgrusewski 04b285486e fix: migrate 4 DQN/recovery test files to GPU types — 92 errors fixed
recovery_tests: Mamba2SSM forward_with_gradients+backward+optimizer_step
gpu_kernel_parity: collect_experiences_gpu, store() not vars(), CudaSlice readback
dqn_gradient_collapse: GpuTensor::randn+to_dtype, host-side gather
dqn_diagnostic: GpuTensor::from_host, to_host for normalization check
trainable_adapter: MlDevice import gated #[cfg(test)]

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-03-19 08:55:55 +01:00

900 lines
27 KiB
Rust

#![allow(
clippy::assertions_on_constants,
clippy::assertions_on_result_states,
clippy::clone_on_copy,
clippy::decimal_literal_representation,
clippy::doc_markdown,
clippy::empty_line_after_doc_comments,
clippy::field_reassign_with_default,
clippy::get_unwrap,
clippy::identity_op,
clippy::inconsistent_digit_grouping,
clippy::indexing_slicing,
clippy::integer_division,
clippy::len_zero,
clippy::let_underscore_must_use,
clippy::manual_div_ceil,
clippy::manual_let_else,
clippy::manual_range_contains,
clippy::modulo_arithmetic,
clippy::needless_range_loop,
clippy::non_ascii_literal,
clippy::redundant_clone,
clippy::shadow_reuse,
clippy::shadow_same,
clippy::shadow_unrelated,
clippy::single_match_else,
clippy::str_to_string,
clippy::string_slice,
clippy::tests_outside_test_module,
clippy::too_many_lines,
clippy::unnecessary_wraps,
clippy::unseparated_literal_suffix,
clippy::use_debug,
clippy::useless_vec,
clippy::wildcard_enum_match_arm,
clippy::else_if_without_else,
clippy::expect_used,
clippy::missing_const_for_fn,
clippy::similar_names,
clippy::type_complexity,
clippy::collapsible_else_if,
clippy::doc_lazy_continuation,
clippy::items_after_test_module,
clippy::map_clone,
clippy::multiple_unsafe_ops_per_block,
clippy::unwrap_or_default,
clippy::assign_op_pattern,
clippy::needless_borrow,
clippy::println_empty_string,
clippy::unnecessary_cast,
clippy::used_underscore_binding,
clippy::create_dir,
clippy::implicit_saturating_sub,
clippy::exit,
clippy::expect_fun_call,
clippy::too_many_arguments,
clippy::unnecessary_map_or,
clippy::unwrap_used,
dead_code,
unused_imports,
unused_variables,
clippy::cloned_ref_to_slice_refs,
clippy::neg_multiply,
clippy::while_let_loop,
clippy::bool_assert_comparison,
clippy::excessive_precision,
clippy::trivially_copy_pass_by_ref,
clippy::op_ref,
clippy::redundant_closure,
clippy::unnecessary_lazy_evaluations,
clippy::if_then_some_else_none,
clippy::unnecessary_to_owned,
clippy::single_component_path_imports,
)]
//! Recovery and Resilience Tests
//!
//! Comprehensive test suite for validating system recovery from failures:
//! Checkpoint corruption, service crashes, OOM errors, GPU failures, and more.
//!
//! # Test Coverage
//!
//! 1. **Checkpoint Recovery** (4 scenarios)
//! - Corruption detection and recovery
//! - Partial checkpoint writes
//! - Metadata corruption
//! - Multi-checkpoint recovery strategy
//!
//! 2. **Service Crash Recovery** (3 scenarios)
//! - Mid-training crash and resume
//! - Multi-job crash recovery
//! - State persistence across restarts
//!
//! 3. **Resource Exhaustion** (3 scenarios)
//! - OOM handling and graceful degradation
//! - GPU memory overflow detection
//! - Disk space exhaustion
//!
//! 4. **Network Failures** (2 scenarios)
//! - Data loading interruption
//! - Checkpoint upload failures
//!
//! # Usage
//!
//! ```bash
//! cargo test -p ml recovery -- --nocapture
//! ```
use anyhow::Result;
use ml_core::cuda_autograd::stream_ops::StreamTensor;
use ml_core::device::MlDevice;
use std::fs;
use std::path::PathBuf;
use std::sync::Arc;
use tempfile::TempDir;
use tracing::{info, warn};
use ml::mamba::{Mamba2Config, Mamba2SSM};
// ============================================================================
// Test Helpers
// ============================================================================
fn create_checkpoint_dir() -> Result<TempDir> {
Ok(TempDir::new()?)
}
fn create_test_config() -> Mamba2Config {
Mamba2Config {
d_model: 64,
d_state: 16,
num_layers: 2,
batch_size: 8,
seq_len: 30,
learning_rate: 1e-4,
..Default::default()
}
}
/// Create a CUDA stream for tests, panics if CUDA is unavailable.
fn test_cuda_stream() -> Arc<cudarc::driver::CudaStream> {
let device = MlDevice::new_cuda(0).expect("CUDA required");
device.cuda_stream().expect("CUDA stream").clone()
}
/// Corrupt a checkpoint file by truncating it
fn corrupt_checkpoint_truncate(path: &PathBuf) -> Result<()> {
fs::write(path, b"TRUNCATED")?;
Ok(())
}
/// Corrupt a checkpoint file by overwriting header
fn corrupt_checkpoint_header(path: &PathBuf) -> Result<()> {
let mut data = fs::read(path)?;
if data.len() > 10 {
// Corrupt first 10 bytes
for byte in data.iter_mut().take(10) {
*byte = 0xFF;
}
fs::write(path, data)?;
}
Ok(())
}
/// Run one training step on a Mamba2SSM model with given input/target.
/// Returns the scalar loss value.
fn train_step(model: &mut Mamba2SSM, input: &StreamTensor, target: &StreamTensor) -> Result<f32> {
model.zero_gradients()?;
let output = model.forward_with_gradients(input)?;
let loss = model.compute_loss(&output, target)?;
let loss_vec = loss.to_vec()?; // gpu-exit: test-only readback
let loss_val = loss_vec.first().copied().unwrap_or(0.0);
model.backward_pass(&loss, input, target)?;
model.optimizer_step()?;
Ok(loss_val)
}
// ============================================================================
// 1. Checkpoint Recovery (4 scenarios)
// ============================================================================
#[tokio::test]
async fn test_checkpoint_corruption_detection_and_recovery() -> Result<()> {
info!("Test: Checkpoint Corruption Detection and Recovery");
let stream = test_cuda_stream();
let config = create_test_config();
let mut model = Mamba2SSM::new(config.clone(), &stream)?;
model.initialize_optimizer()?;
let checkpoint_dir = create_checkpoint_dir()?;
// Save checkpoint v1
let checkpoint_v1 = checkpoint_dir.path().join("checkpoint_v1.safetensors");
info!("Saving checkpoint v1");
model
.save_checkpoint(checkpoint_v1.to_str().unwrap())
.await?;
let v1_size = fs::metadata(&checkpoint_v1)?.len();
info!(bytes = v1_size, "Checkpoint v1 saved");
// Train a bit more
let input = StreamTensor::randn(&[8, 30, 64], 1.0, &stream)?;
let target = StreamTensor::randn(&[8, 1], 1.0, &stream)?;
for _ in 0..3 {
train_step(&mut model, &input, &target)?;
}
// Save checkpoint v2
let checkpoint_v2 = checkpoint_dir.path().join("checkpoint_v2.safetensors");
info!("Saving checkpoint v2");
model
.save_checkpoint(checkpoint_v2.to_str().unwrap())
.await?;
let v2_size = fs::metadata(&checkpoint_v2)?.len();
info!(bytes = v2_size, "Checkpoint v2 saved");
// Corrupt v2
info!("Corrupting checkpoint v2");
corrupt_checkpoint_truncate(&checkpoint_v2)?;
let v2_corrupted_size = fs::metadata(&checkpoint_v2)?.len();
info!(bytes = v2_corrupted_size, "Checkpoint v2 corrupted");
// Try to load v2 (should fail)
info!("Attempting to load corrupted v2");
let result = model.load_checkpoint(checkpoint_v2.to_str().unwrap()).await;
assert!(result.is_err(), "Should detect corruption in v2");
info!("Corruption detected");
// Fallback to v1
info!("Falling back to v1");
model
.load_checkpoint(checkpoint_v1.to_str().unwrap())
.await?;
info!("Recovered from v1");
// Verify model works
let output = model.forward(&input)?;
assert!(output.shape[0] == 8, "Model should work after recovery");
info!("Model operational after recovery");
info!("Checkpoint recovery test PASSED");
Ok(())
}
#[tokio::test]
async fn test_partial_checkpoint_write() -> Result<()> {
info!("Test: Partial Checkpoint Write Detection");
let stream = test_cuda_stream();
let config = create_test_config();
let mut model = Mamba2SSM::new(config.clone(), &stream)?;
model.initialize_optimizer()?;
let checkpoint_dir = create_checkpoint_dir()?;
let full_checkpoint = checkpoint_dir.path().join("full.safetensors");
// Save complete checkpoint
info!("Saving complete checkpoint");
model
.save_checkpoint(full_checkpoint.to_str().unwrap())
.await?;
let full_size = fs::metadata(&full_checkpoint)?.len();
info!(bytes = full_size, "Full checkpoint saved");
// Create partial checkpoint (50% of size)
let partial_checkpoint = checkpoint_dir.path().join("partial.safetensors");
let data = fs::read(&full_checkpoint)?;
let partial_data = &data[..data.len() / 2];
fs::write(&partial_checkpoint, partial_data)?;
let partial_size = fs::metadata(&partial_checkpoint)?.len();
info!(bytes = partial_size, pct = (partial_size as f64 / full_size as f64) * 100.0, "Created partial checkpoint");
// Try to load partial checkpoint
info!("Attempting to load partial checkpoint");
let result = model
.load_checkpoint(partial_checkpoint.to_str().unwrap())
.await;
assert!(result.is_err(), "Should detect incomplete checkpoint");
info!("Incomplete checkpoint detected");
// Verify full checkpoint still works
info!("Loading full checkpoint");
model
.load_checkpoint(full_checkpoint.to_str().unwrap())
.await?;
info!("Full checkpoint loaded successfully");
info!("Partial checkpoint detection test PASSED");
Ok(())
}
#[tokio::test]
async fn test_metadata_corruption() -> Result<()> {
info!("Test: Checkpoint Metadata Corruption");
let stream = test_cuda_stream();
let config = create_test_config();
let mut model = Mamba2SSM::new(config.clone(), &stream)?;
model.initialize_optimizer()?;
let checkpoint_dir = create_checkpoint_dir()?;
let checkpoint_path = checkpoint_dir.path().join("test.safetensors");
// Save checkpoint
info!("Saving checkpoint");
model
.save_checkpoint(checkpoint_path.to_str().unwrap())
.await?;
info!("Checkpoint saved");
// Corrupt header/metadata
info!("Corrupting checkpoint header");
corrupt_checkpoint_header(&checkpoint_path)?;
info!("Header corrupted");
// Try to load
info!("Attempting to load corrupted checkpoint");
let result = model
.load_checkpoint(checkpoint_path.to_str().unwrap())
.await;
assert!(result.is_err(), "Should detect header corruption");
info!("Header corruption detected");
info!("Metadata corruption test PASSED");
Ok(())
}
#[tokio::test]
async fn test_multi_checkpoint_recovery_strategy() -> Result<()> {
info!("Test: Multi-Checkpoint Recovery Strategy");
let stream = test_cuda_stream();
let config = create_test_config();
let mut model = Mamba2SSM::new(config.clone(), &stream)?;
model.initialize_optimizer()?;
let checkpoint_dir = create_checkpoint_dir()?;
// Create 5 checkpoints
let mut checkpoints = Vec::new();
info!("Creating checkpoints");
for i in 1..=5 {
let path = checkpoint_dir
.path()
.join(format!("checkpoint_{}.safetensors", i));
model.save_checkpoint(path.to_str().unwrap()).await?;
checkpoints.push(path);
info!(checkpoint = i, "Checkpoint saved");
// Train a bit between checkpoints
let input = StreamTensor::randn(&[8, 30, 64], 1.0, &stream)?;
let target = StreamTensor::randn(&[8, 1], 1.0, &stream)?;
train_step(&mut model, &input, &target)?;
}
// Corrupt checkpoints 3, 4, 5
info!("Corrupting checkpoints 3, 4, 5");
for i in 3..=5 {
corrupt_checkpoint_truncate(&checkpoints[i - 1])?;
info!(checkpoint = i, "Checkpoint corrupted");
}
// Recovery strategy: try from newest to oldest
info!("Attempting recovery (newest to oldest)");
let mut recovered = false;
for (idx, checkpoint) in checkpoints.iter().enumerate().rev() {
let checkpoint_num = idx + 1;
match model.load_checkpoint(checkpoint.to_str().unwrap()).await {
Ok(_) => {
info!(checkpoint = checkpoint_num, "Recovery succeeded");
recovered = true;
assert!(checkpoint_num <= 2, "Should recover from checkpoint 1 or 2");
break;
},
Err(e) => {
warn!(checkpoint = checkpoint_num, error = %e, "Checkpoint load failed");
},
}
}
assert!(recovered, "Should recover from at least one checkpoint");
info!("Successfully recovered from valid checkpoint");
info!("Multi-checkpoint recovery test PASSED");
Ok(())
}
// ============================================================================
// 2. Service Crash Recovery (3 scenarios)
// ============================================================================
#[tokio::test]
async fn test_mid_training_crash_and_resume() -> Result<()> {
info!("Test: Mid-Training Crash and Resume");
let stream = test_cuda_stream();
let checkpoint_dir = create_checkpoint_dir()?;
// Training state
#[derive(Debug, Clone)]
struct TrainingState {
epoch: usize,
total_epochs: usize,
last_loss: f32,
checkpoint_path: Option<String>,
}
let mut state = TrainingState {
epoch: 0,
total_epochs: 10,
last_loss: 1.0,
checkpoint_path: None,
};
info!("Phase 1: Initial training (until crash)");
let config = create_test_config();
let mut model = Mamba2SSM::new(config.clone(), &stream)?;
model.initialize_optimizer()?;
let input = StreamTensor::randn(&[8, 30, 64], 1.0, &stream)?;
let target = StreamTensor::randn(&[8, 1], 1.0, &stream)?;
// Train for 4 epochs, then "crash"
for epoch in 0..4 {
let loss_val = train_step(&mut model, &input, &target)?;
state.last_loss = loss_val;
state.epoch = epoch + 1;
info!(epoch = state.epoch, total = state.total_epochs, loss = state.last_loss, "Training epoch");
// Save checkpoint every epoch
let checkpoint_path = checkpoint_dir
.path()
.join(format!("epoch_{}.safetensors", state.epoch));
model
.save_checkpoint(checkpoint_path.to_str().unwrap())
.await?;
state.checkpoint_path = Some(checkpoint_path.to_string_lossy().to_string());
}
warn!(epoch = state.epoch, "CRASH: Service terminated");
// Drop model (simulate crash)
drop(model);
// Phase 2: Resume from checkpoint
info!(epoch = state.epoch, "Phase 2: Service restart and resume");
let mut resumed_model = Mamba2SSM::new(config, &stream)?;
resumed_model.initialize_optimizer()?;
resumed_model
.load_checkpoint(state.checkpoint_path.as_ref().unwrap())
.await?;
info!("Checkpoint loaded");
// Continue training
info!("Continuing training");
for epoch in state.epoch..state.total_epochs {
let loss_val = train_step(&mut resumed_model, &input, &target)?;
info!(epoch = epoch + 1, total = state.total_epochs, loss = loss_val, "Resume epoch");
}
info!("Training completed after recovery");
info!("Mid-training crash recovery test PASSED");
Ok(())
}
#[tokio::test]
async fn test_multi_job_crash_recovery() -> Result<()> {
info!("Test: Multi-Job Crash Recovery");
let stream = test_cuda_stream();
#[derive(Debug, Clone)]
struct Job {
id: String,
progress: f32,
checkpoint: Option<String>,
}
let checkpoint_dir = create_checkpoint_dir()?;
// Create 3 jobs
let mut jobs = vec![
Job {
id: "job_1".to_string(),
progress: 0.0,
checkpoint: None,
},
Job {
id: "job_2".to_string(),
progress: 0.0,
checkpoint: None,
},
Job {
id: "job_3".to_string(),
progress: 0.0,
checkpoint: None,
},
];
info!(job_count = jobs.len(), "Phase 1: Running jobs");
// Run each job partially
for job in jobs.iter_mut() {
info!(job_id = %job.id, "Processing job");
let config = create_test_config();
let mut model = Mamba2SSM::new(config, &stream)?;
model.initialize_optimizer()?;
let input = StreamTensor::randn(&[8, 30, 64], 1.0, &stream)?;
let target = StreamTensor::randn(&[8, 1], 1.0, &stream)?;
// Train for 2 steps
for step in 0..2 {
train_step(&mut model, &input, &target)?;
job.progress = (step + 1) as f32 / 5.0; // 5 total steps
}
// Save checkpoint
let checkpoint_path = checkpoint_dir
.path()
.join(format!("{}.safetensors", job.id));
model
.save_checkpoint(checkpoint_path.to_str().unwrap())
.await?;
job.checkpoint = Some(checkpoint_path.to_string_lossy().to_string());
info!(progress_pct = job.progress * 100.0, "Checkpoint saved");
}
warn!("CRASH: All jobs interrupted");
// Phase 2: Recover all jobs
info!(job_count = jobs.len(), "Phase 2: Recovering jobs");
let mut recovered_count = 0;
for job in jobs.iter() {
info!(job_id = %job.id, "Recovering job");
let config = create_test_config();
let mut model = Mamba2SSM::new(config, &stream)?;
model.initialize_optimizer()?;
if let Some(checkpoint) = &job.checkpoint {
match model.load_checkpoint(checkpoint).await {
Ok(_) => {
info!(job_id = %job.id, progress_pct = job.progress * 100.0, "Job recovered");
recovered_count += 1;
},
Err(e) => {
warn!(job_id = %job.id, error = %e, "Job recovery failed");
},
}
}
}
info!(recovered = recovered_count, total = jobs.len(), "Job recovery complete");
assert_eq!(recovered_count, jobs.len(), "Should recover all jobs");
info!("Multi-job recovery test PASSED");
Ok(())
}
#[tokio::test]
async fn test_state_persistence_across_restarts() -> Result<()> {
info!("Test: State Persistence Across Restarts");
let stream = test_cuda_stream();
let checkpoint_dir = create_checkpoint_dir()?;
let checkpoint_path = checkpoint_dir.path().join("persistent.safetensors");
let input = StreamTensor::randn(&[8, 30, 64], 1.0, &stream)?;
let target = StreamTensor::randn(&[8, 1], 1.0, &stream)?;
let mut losses = Vec::new();
// Restart 1: Initial training
info!("Restart 1: Initial training");
{
let config = create_test_config();
let mut model = Mamba2SSM::new(config, &stream)?;
model.initialize_optimizer()?;
for _ in 0..2 {
let loss_val = train_step(&mut model, &input, &target)?;
losses.push(loss_val);
}
model
.save_checkpoint(checkpoint_path.to_str().unwrap())
.await?;
info!(loss = losses.last().copied().unwrap_or(0.0), "Restart 1 loss");
}
// Restart 2: Resume and continue
info!("Restart 2: Resume training");
{
let config = create_test_config();
let mut model = Mamba2SSM::new(config, &stream)?;
model.initialize_optimizer()?;
model
.load_checkpoint(checkpoint_path.to_str().unwrap())
.await?;
for _ in 0..2 {
let loss_val = train_step(&mut model, &input, &target)?;
losses.push(loss_val);
}
model
.save_checkpoint(checkpoint_path.to_str().unwrap())
.await?;
info!(loss = losses.last().copied().unwrap_or(0.0), "Restart 2 loss");
}
// Restart 3: Final resume
info!("Restart 3: Final resume");
{
let config = create_test_config();
let mut model = Mamba2SSM::new(config, &stream)?;
model.initialize_optimizer()?;
model
.load_checkpoint(checkpoint_path.to_str().unwrap())
.await?;
for _ in 0..2 {
let loss_val = train_step(&mut model, &input, &target)?;
losses.push(loss_val);
}
info!(loss = losses.last().copied().unwrap_or(0.0), "Restart 3 loss");
}
info!(losses = ?losses, restarts = 3, "State persisted across restarts");
// Validate losses are monotonically decreasing (or at least not increasing significantly)
assert!(losses.len() == 6, "Should have 6 training steps");
info!("State persistence test PASSED");
Ok(())
}
// ============================================================================
// 3. Resource Exhaustion (3 scenarios)
// ============================================================================
#[tokio::test]
async fn test_oom_handling_graceful_degradation() -> Result<()> {
info!("Test: OOM Handling and Graceful Degradation");
let stream = test_cuda_stream();
let config = create_test_config();
// Start with large batch size
let mut batch_size = 128usize;
let min_batch_size = 8usize;
info!(start = batch_size, min = min_batch_size, "Testing batch sizes");
while batch_size >= min_batch_size {
info!(batch_size, "Trying batch size");
let mut test_config = config.clone();
test_config.batch_size = batch_size;
let mut model = Mamba2SSM::new(test_config, &stream)?;
model.initialize_optimizer()?;
// Try to allocate and train
let result = (|| -> Result<()> {
let input = StreamTensor::randn(&[batch_size, 30, 64], 1.0, &stream)?;
let target = StreamTensor::randn(&[batch_size, 1], 1.0, &stream)?;
train_step(&mut model, &input, &target)?;
Ok(())
})();
match result {
Ok(_) => {
info!(batch_size, "Batch size succeeded");
break; // Found working batch size
},
Err(e) => {
warn!(batch_size, error = %e, "Batch size failed");
// Reduce batch size by half
batch_size /= 2;
if batch_size < min_batch_size {
warn!("Could not find working batch size");
break;
}
info!(batch_size, "Degrading batch size");
},
}
}
assert!(
batch_size >= min_batch_size,
"Should find working batch size"
);
info!(batch_size, "Gracefully degraded to batch size");
info!("OOM handling test PASSED");
Ok(())
}
#[tokio::test]
async fn test_gpu_memory_overflow_detection() -> Result<()> {
info!("Test: GPU Memory Overflow Detection");
let device = MlDevice::new_cuda(0).expect("CUDA required");
let stream = device.cuda_stream().expect("CUDA stream").clone();
if !device.is_cuda() {
warn!("Skipping: CUDA not available");
return Ok(());
}
info!(device = ?device, "Device selected");
// Try to allocate increasingly large tensors
let mut allocated_mb = 0.0f64;
let increment_mb = 100.0; // 100 MB increments
info!(increment_mb, "Allocating tensors in MB increments");
for _i in 1..=50 {
let elements = (increment_mb * 1024.0 * 1024.0 / 4.0) as usize; // 4 bytes per f32
let result = StreamTensor::zeros(&[elements], &stream);
match result {
Ok(_tensor) => {
allocated_mb += increment_mb;
info!(allocated_mb, "Allocation succeeded");
},
Err(e) => {
info!(allocated_mb, error = %e, "GPU memory limit detected");
break;
},
}
}
info!("GPU memory overflow detection works");
info!("GPU memory overflow test PASSED");
Ok(())
}
#[tokio::test]
async fn test_disk_space_exhaustion() -> Result<()> {
info!("Test: Disk Space Exhaustion Detection");
let stream = test_cuda_stream();
let config = create_test_config();
let mut model = Mamba2SSM::new(config, &stream)?;
model.initialize_optimizer()?;
let checkpoint_dir = create_checkpoint_dir()?;
// Save a checkpoint to measure size
let test_checkpoint = checkpoint_dir.path().join("test.safetensors");
model
.save_checkpoint(test_checkpoint.to_str().unwrap())
.await?;
let checkpoint_size = fs::metadata(&test_checkpoint)?.len();
info!(bytes = checkpoint_size, mb = checkpoint_size as f64 / 1024.0 / 1024.0, "Checkpoint size");
info!("Checkpoint saved successfully");
// Simulate insufficient space by trying to write to a location that doesn't exist
let invalid_path = PathBuf::from("/nonexistent/directory/checkpoint.safetensors");
let result = model.save_checkpoint(invalid_path.to_str().unwrap()).await;
match result {
Ok(_) => {
panic!("Should not succeed writing to invalid path");
},
Err(e) => {
info!(error = %e, "Invalid path correctly rejected");
},
}
info!("Disk space issues can be detected");
info!("Disk space exhaustion test PASSED");
Ok(())
}
// ============================================================================
// 4. Network Failures (2 scenarios)
// ============================================================================
#[tokio::test]
async fn test_data_loading_interruption() -> Result<()> {
info!("Test: Data Loading Interruption");
// Simulate data loading from non-existent source
let invalid_path = PathBuf::from("/nonexistent/data/file.dbn.zst");
info!(path = ?invalid_path, "Attempting to load from invalid path");
// This would normally use DbnSequenceLoader, but we'll simulate the error
let result: Result<()> = if invalid_path.exists() {
Ok(())
} else {
Err(anyhow::anyhow!("Data file not found: {:?}", invalid_path))
};
match result {
Ok(_) => {
panic!("Should fail with non-existent path");
},
Err(e) => {
info!(error = %e, "Error detected for invalid path");
},
}
info!("Data loading interruption handled gracefully");
info!("Data loading interruption test PASSED");
Ok(())
}
#[tokio::test]
async fn test_checkpoint_upload_failures() -> Result<()> {
info!("Test: Checkpoint Upload Failures");
let stream = test_cuda_stream();
let config = create_test_config();
let mut model = Mamba2SSM::new(config, &stream)?;
model.initialize_optimizer()?;
// Try to save to invalid location
let invalid_path = PathBuf::from("/root/protected/checkpoint.safetensors");
info!(path = ?invalid_path, "Attempting to save to protected location");
let result = model.save_checkpoint(invalid_path.to_str().unwrap()).await;
match result {
Ok(_) => {
warn!("Save to protected path succeeded (may have permissions)");
},
Err(e) => {
info!(error = %e, "Protected path correctly rejected");
},
}
// Try to save to valid location (should succeed)
let checkpoint_dir = create_checkpoint_dir()?;
let valid_path = checkpoint_dir.path().join("valid.safetensors");
info!("Attempting to save to valid location");
model.save_checkpoint(valid_path.to_str().unwrap()).await?;
info!("Save to valid location succeeded");
info!("Checkpoint upload failures can be detected");
info!("Checkpoint upload failure test PASSED");
Ok(())
}
// ============================================================================
// Test Summary
// ============================================================================
#[tokio::test]
async fn test_recovery_summary() -> Result<()> {
info!("Recovery and Resilience Test Summary: 12 scenarios across checkpoint/crash/resource/network categories");
Ok(())
}