wireshift-uring 0.1.1

#![allow(missing_docs)]
#![allow(clippy::unwrap_used)]

use std::fs::File;
use std::os::fd::{AsRawFd, FromRawFd};
use std::sync::{Arc, Barrier};
use std::thread;
use std::time::Duration;
use tempfile::NamedTempFile;

use wireshift_core::backend::{Backend, BackendSubmission, CancellationHandle};
use wireshift_core::config::RingConfig;
use wireshift_core::op::OpDescriptor;
use wireshift_uring::UringBackend;

#[test]
fn test_edge_queue_depth_exhaustion() {
    let (tx, rx) = std::sync::mpsc::channel();
    let config = RingConfig {
        queue_depth: 8, // Very small queue depth
        ..RingConfig::default()
    };
    let backend = UringBackend::new(&config, tx).expect("backend creation");

    // Submit 32 operations rapidly to exhaust the queue depth.
    // Some should be accepted, some might get rejected or wait in pending.
    // The key invariant is that it must not crash, panic, or deadlock.
    let mut submitted_count = 0;
    for i in 0..32 {
        let submission = BackendSubmission {
            timeout: None,
            id: 1000 + i,
            descriptor: OpDescriptor::Nop,
        };
        if backend.submit(submission).is_ok() {
            submitted_count += 1;
        }
    }

    assert!(submitted_count > 0, "At least some submissions should be accepted");

    // Let them process a bit
    thread::sleep(Duration::from_millis(50));

    // Drain the completions channel to make sure we don't leak/block
    while rx.try_recv().is_ok() {}

    // Shutdown backend safely
    let _ = backend.shutdown();
}

#[test]
fn test_edge_submit_and_wait_timeout_boundary() {
    let (tx, _rx) = std::sync::mpsc::channel();
    let config = RingConfig {
        queue_depth: 16,
        ..RingConfig::default()
    };
    let backend = UringBackend::new(&config, tx).expect("backend creation");

    // Create a pipe using libc so we can have a stuck read.
    // A read from a pipe with no writer will block in the kernel, making it a perfect stuck operation.
    let mut fds = [0 as libc::c_int; 2];
    let res = unsafe { libc::pipe(fds.as_mut_ptr()) };
    assert_eq!(res, 0, "failed to create pipe");

    let read_file = unsafe { File::from_raw_fd(fds[0]) };
    let _write_file = unsafe { File::from_raw_fd(fds[1]) };

    // Submit a stuck read operation
    let buffer = wireshift_core::buffer::BufferPool::new(1024, 1)
        .unwrap()
        .acquire()
        .unwrap()
        .into_submitted();

    let submission = BackendSubmission {
        timeout: None,
        id: 2000,
        descriptor: OpDescriptor::Read {
            file: read_file,
            offset: 0,
            buffer,
            len: None,
        },
    };

    backend.submit(submission).expect("submit stuck operation");

    // Give it a tiny moment to start in the manager thread
    thread::sleep(Duration::from_millis(20));

    // Shutdown backend. Since we have a stuck read, this will trigger the manager's
    // cancel-on-shutdown logic. Since AsyncCancel might not complete instantly
    // (or if it gets stuck), the manager's 5-second shutdown timeout boundary is tested!
    // It should exit and join without hanging indefinitely.
    let start = std::time::Instant::now();
    let shutdown_res = backend.shutdown();
    let elapsed = start.elapsed();

    // Verify shutdown completed successfully and within a reasonable boundary (should be quick because cancel is fast, but must definitely be < 10 seconds)
    assert!(shutdown_res.is_ok(), "shutdown should succeed");
    assert!(elapsed < Duration::from_secs(10), "shutdown took too long: {:?}", elapsed);
}

#[test]
fn test_edge_raw_fd_registrations() {
    let (tx, _rx) = std::sync::mpsc::channel();
    let config = RingConfig {
        queue_depth: 32,
        ..RingConfig::default()
    };
    let backend = UringBackend::new(&config, tx).expect("backend creation");

    // 1. Unregistering when nothing is registered should be a safe no-op or succeed
    let _ = backend.unregister_files();

    // 2. Registering an empty array
    let res_empty = backend.register_files(&[]);
    assert!(res_empty.is_err(), "register empty files should fail in kernel");

    // 3. Registering valid file descriptors
    let file1 = File::open("/dev/null").expect("open file1");
    let file2 = File::open("/dev/null").expect("open file2");
    let fds = [file1.as_raw_fd(), file2.as_raw_fd()];
    
    let res_reg = backend.register_files(&fds);
    assert!(res_reg.is_ok(), "registering valid fds should succeed");

    // 4. Double registration should handle gracefully
    // (either succeed or return error, but not crash/panic/leak)
    let _ = backend.register_files(&fds);

    // 5. Registering invalid file descriptor
    let invalid_fds = [-1, 99999];
    let res_invalid = backend.register_files(&invalid_fds);
    // Should fail cleanly in the kernel/backend without panicking
    assert!(res_invalid.is_err(), "registering invalid fds must fail");

    // 6. Clean up
    let _ = backend.unregister_files();
    let _ = backend.shutdown();
}

#[test]
fn test_edge_concurrent_submissions() {
    let (tx, rx) = std::sync::mpsc::channel();
    let config = RingConfig {
        queue_depth: 128,
        ..RingConfig::default()
    };
    let backend = Arc::new(UringBackend::new(&config, tx).expect("backend creation"));

    let num_threads = 8;
    let ops_per_thread = 100;
    let barrier = Arc::new(Barrier::new(num_threads));
    let mut handles = Vec::new();

    for t_idx in 0..num_threads {
        let backend_clone = Arc::clone(&backend);
        let barrier_clone = Arc::clone(&barrier);

        handles.push(thread::spawn(move || {
            barrier_clone.wait();
            for i in 0..ops_per_thread {
                let id = (t_idx as u64 * 1000) + i as u64;
                let submission = BackendSubmission {
                    timeout: None,
                    id,
                    descriptor: OpDescriptor::Nop,
                };
                let _ = backend_clone.submit(submission);

                // Occasionally submit cancellations concurrently
                if i % 10 == 0 {
                    let cancel_handle = CancellationHandle { target: id };
                    let _ = backend_clone.cancel(cancel_handle);
                }
            }
        }));
    }

    for h in handles {
        h.join().expect("thread join succeeded");
    }

    // Let the manager process completions
    thread::sleep(Duration::from_millis(50));

    // Drain all completions
    while rx.try_recv().is_ok() {}

    // Shutdown backend and make sure it shuts down cleanly under concurrent stress
    let _ = backend.shutdown();
}