wireshift-fallback 0.1.1

use std::sync::mpsc;
use std::sync::Arc;
use std::thread;

use wireshift_core::backend::{Backend, BackendSubmission, CancellationHandle};
use wireshift_core::config::RingConfig;
use wireshift_core::op::OpDescriptor;
use wireshift_fallback::FallbackBackend;

#[test]
fn test_depth_concurrent_32_threads_hammer_api() {
    let (tx, rx) = mpsc::channel();

    let config = RingConfig::default()
        .with_fallback_threads(32)
        .with_queue_depth(1024); // Large queue to prevent blocking sends on TrySendError::Full during 32 thread hammer

    let backend = Arc::new(FallbackBackend::new(&config, tx).unwrap());

    let num_threads = 32;
    let ops_per_thread = 1_000;

    let mut join_handles = vec![];

    for thread_idx in 0..num_threads {
        let backend_clone = backend.clone();

        let handle = thread::spawn(move || {
            for op_idx in 0..ops_per_thread {
                let id = thread_idx * ops_per_thread + op_idx;

                let mut retry_count = 0;
                loop {
                    let submission = BackendSubmission { id, descriptor: OpDescriptor::Nop, timeout: None };
                    let result = backend_clone.submit(submission);
                    match result {
                        Ok(_) => break,
                        Err(e) => {
                            if format!("{:?}", e).contains("queue is full") {
                                // Simple exponential backoff for full queues
                                retry_count += 1;
                                assert!(retry_count < 100, "Queue remained full for too long");
                                thread::yield_now();
                            } else {
                                panic!("Unexpected submit error: {:?}", e);
                            }
                        }
                    }
                }

                // Immediately attempt cancellation on every 10th op to test race conditions
                // in CancelRegistry between worker execution and cancellation request.
                if id % 10 == 0 {
                    let cancellation = CancellationHandle { target: id };
                    // Ignore the result as it could be either canceled or already completed/missing.
                    // This explicitly checks that `cancel` does not panic under heavy load.
                    let _cancel_result = backend_clone.cancel(cancellation);
                }
            }
        });

        join_handles.push(handle);
    }

    for handle in join_handles {
        handle.join().unwrap();
    }

    backend.shutdown().unwrap();

    let total_expected = num_threads * ops_per_thread;
    let mut completed = 0;

    while let Ok(completion) = rx.try_recv() {
        completed += 1;
        // Verify results
        match completion.result {
            Ok(payload) => assert!(
                matches!(payload, wireshift_core::op::CompletionPayload::Unit),
                "Expected Nop payload"
            ),
            Err(e) => {
                // Assert it's an expected cancellation error
                assert!(
                    format!("{:?}", e).contains("canceled before"),
                    "Only cancellation errors expected, got {:?}",
                    e
                );
            }
        }
    }

    assert_eq!(
        completed, total_expected,
        "Not all operations completed or returned a cancellation payload"
    );
}