freenet 0.2.43

//! Tests for the Op Request Mediator crash and recovery scenarios.
//!
//! These tests verify:
//! - Graceful handling of executor drops
//! - Stale request cleanup
//! - Channel closure behavior
//! - Capacity limits and backpressure

use crate::config::GlobalExecutor;
use std::sync::Arc;
use std::sync::atomic::{AtomicU32, Ordering};
use std::time::Duration;

use tokio::sync::{mpsc, oneshot};
use tokio::time::timeout;

use crate::contract::executor::{OpRequestError, OpRequestReceiver, OpRequestSender};
use crate::message::Transaction;
use crate::operations::OpEnum;
use crate::operations::get::GetMsg;

// =============================================================================
// Test Helper: Simplified Mediator for Unit Testing
// =============================================================================

/// A simplified mediator that mirrors the real mediator's core behavior.
/// Used for testing without requiring the full event loop infrastructure.
struct TestMediator {
    op_request_rx: OpRequestReceiver,
    to_event_loop_tx: mpsc::Sender<Transaction>,
    from_event_loop_rx: mpsc::Receiver<OpEnum>,
    pending:
        std::collections::HashMap<Transaction, oneshot::Sender<Result<OpEnum, OpRequestError>>>,
    max_pending: usize,
}

impl TestMediator {
    fn new(
        op_request_rx: OpRequestReceiver,
        to_event_loop_tx: mpsc::Sender<Transaction>,
        from_event_loop_rx: mpsc::Receiver<OpEnum>,
    ) -> Self {
        Self {
            op_request_rx,
            to_event_loop_tx,
            from_event_loop_rx,
            pending: std::collections::HashMap::new(),
            max_pending: 100, // Lower limit for testing
        }
    }

    async fn process_one_request(&mut self) -> bool {
        tokio::select! {
            Some((tx, response_tx)) = self.op_request_rx.recv() => {
                // Check capacity
                if self.pending.len() >= self.max_pending {
                    drop(response_tx.send(Err(OpRequestError::Failed(
                        "mediator at capacity".to_string()
                    ))));
                    return true;
                }

                // Store pending and forward to event loop
                self.pending.insert(tx, response_tx);
                if self.to_event_loop_tx.send(tx).await.is_err() {
                    if let Some(pending) = self.pending.remove(&tx) {
                        drop(pending.send(Err(OpRequestError::ChannelClosed)));
                    }
                }
                true
            }
            Some(op_result) = self.from_event_loop_rx.recv() => {
                let tx = *op_result.id();
                if let Some(pending) = self.pending.remove(&tx) {
                    drop(pending.send(Ok(op_result)));
                }
                true
            }
            else => false
        }
    }

    fn pending_count(&self) -> usize {
        self.pending.len()
    }
}

// =============================================================================
// Channel Creation Helper
// =============================================================================

fn create_test_channels() -> (
    OpRequestSender,
    OpRequestReceiver,
    mpsc::Sender<Transaction>,
    mpsc::Receiver<Transaction>,
    mpsc::Sender<OpEnum>,
    mpsc::Receiver<OpEnum>,
) {
    let (op_tx, op_rx) = mpsc::channel(100);
    let (to_el_tx, to_el_rx) = mpsc::channel(100);
    let (from_el_tx, from_el_rx) = mpsc::channel(100);
    (op_tx, op_rx, to_el_tx, to_el_rx, from_el_tx, from_el_rx)
}

// =============================================================================
// Executor Drop Recovery Tests
// =============================================================================

#[tokio::test]
async fn test_executor_drops_before_response() {
    let (op_tx, op_rx, to_el_tx, _to_el_rx, _from_el_tx, from_el_rx) = create_test_channels();
    let mut mediator = TestMediator::new(op_rx, to_el_tx, from_el_rx);

    let tx = Transaction::new::<GetMsg>();

    // Send request but drop the response receiver immediately
    let (response_tx, response_rx) = oneshot::channel();
    op_tx.send((tx, response_tx)).await.unwrap();
    drop(response_rx); // Simulate executor dropping before response

    // Mediator should process the request without panicking
    mediator.process_one_request().await;
    assert_eq!(mediator.pending_count(), 1);
}

#[tokio::test]
async fn test_multiple_executors_drop_independently() {
    let (op_tx, op_rx, to_el_tx, _to_el_rx, _from_el_tx, from_el_rx) = create_test_channels();
    let mut mediator = TestMediator::new(op_rx, to_el_tx, from_el_rx);

    let dropped = Arc::new(AtomicU32::new(0));

    // Send multiple requests, some will be dropped
    for i in 0..5 {
        let tx = Transaction::new::<GetMsg>();
        let (response_tx, response_rx) = oneshot::channel();
        op_tx.send((tx, response_tx)).await.unwrap();

        // Drop odd-numbered requests
        if i % 2 == 1 {
            drop(response_rx);
            dropped.fetch_add(1, Ordering::SeqCst);
        } else {
            // Keep even ones alive briefly then drop
            GlobalExecutor::spawn(async move {
                tokio::time::sleep(Duration::from_millis(100)).await;
                drop(response_rx);
            });
        }
    }

    // Process all requests
    for _ in 0..5 {
        mediator.process_one_request().await;
    }

    // All should be pending (mediator doesn't know about drops until it tries to respond)
    assert_eq!(mediator.pending_count(), 5);
}

// =============================================================================
// Stale Request Cleanup Tests
// =============================================================================

#[tokio::test]
async fn test_stale_requests_identified() {
    use std::time::Instant;

    let mut pending: std::collections::HashMap<
        Transaction,
        (oneshot::Sender<Result<OpEnum, OpRequestError>>, Instant),
    > = std::collections::HashMap::new();

    // Add some requests with different ages
    for i in 0..5 {
        let tx = Transaction::new::<GetMsg>();
        let (response_tx, _response_rx) = oneshot::channel();

        // Simulate different creation times
        let created_at = Instant::now() - Duration::from_secs(i * 60);
        pending.insert(tx, (response_tx, created_at));
    }

    // Find stale requests (older than 2 minutes)
    let threshold = Duration::from_secs(120);
    let stale: Vec<_> = pending
        .iter()
        .filter(|(_, (_, created))| created.elapsed() > threshold)
        .map(|(tx, _)| *tx)
        .collect();

    // Requests 2, 3, 4 should be stale (120s, 180s, 240s old)
    assert_eq!(stale.len(), 3);
}

#[tokio::test]
async fn test_stale_cleanup_notifies_waiters() {
    let (response_tx, response_rx) = oneshot::channel::<Result<OpEnum, OpRequestError>>();

    // Simulate stale cleanup notification
    drop(response_tx.send(Err(OpRequestError::Failed(
        "request exceeded stale threshold".to_string(),
    ))));

    // Waiter should receive the error
    let result = response_rx.await.unwrap();
    assert!(matches!(result, Err(OpRequestError::Failed(msg)) if msg.contains("stale")));
}

// =============================================================================
// Channel Closure Tests
// =============================================================================

#[tokio::test]
async fn test_event_loop_channel_closes() {
    let (op_tx, op_rx, to_el_tx, _to_el_rx, _from_el_tx, from_el_rx) = create_test_channels();

    // Drop the event loop receiver to simulate event loop crash
    drop(_to_el_rx);

    let mut mediator = TestMediator::new(op_rx, to_el_tx, from_el_rx);

    let tx = Transaction::new::<GetMsg>();
    let (response_tx, response_rx) = oneshot::channel();
    op_tx.send((tx, response_tx)).await.unwrap();

    // Process request - should fail to forward
    mediator.process_one_request().await;

    // Waiter should receive channel closed error
    let result = timeout(Duration::from_millis(100), response_rx).await;
    assert!(result.is_ok());
    let response = result.unwrap().unwrap();
    assert!(matches!(response, Err(OpRequestError::ChannelClosed)));
}

#[tokio::test]
async fn test_all_senders_dropped_mediator_exits() {
    let (op_tx, op_rx, to_el_tx, _to_el_rx, _from_el_tx, from_el_rx) = create_test_channels();
    let mut mediator = TestMediator::new(op_rx, to_el_tx, from_el_rx);

    // Drop all senders
    drop(op_tx);
    drop(_from_el_tx);

    // Mediator should detect closure and return false
    let still_running = mediator.process_one_request().await;
    assert!(!still_running);
}

// =============================================================================
// Capacity and Backpressure Tests
// =============================================================================

#[tokio::test]
async fn test_mediator_rejects_at_capacity() {
    let (op_tx, op_rx, to_el_tx, _to_el_rx, _from_el_tx, from_el_rx) = create_test_channels();
    let mut mediator = TestMediator::new(op_rx, to_el_tx, from_el_rx);
    mediator.max_pending = 5; // Low limit for testing

    // Fill up to capacity
    let mut receivers = Vec::new();
    for _ in 0..5 {
        let tx = Transaction::new::<GetMsg>();
        let (response_tx, response_rx) = oneshot::channel();
        op_tx.send((tx, response_tx)).await.unwrap();
        receivers.push(response_rx);
        mediator.process_one_request().await;
    }

    assert_eq!(mediator.pending_count(), 5);

    // Next request should be rejected
    let tx = Transaction::new::<GetMsg>();
    let (response_tx, response_rx) = oneshot::channel();
    op_tx.send((tx, response_tx)).await.unwrap();
    mediator.process_one_request().await;

    // Check that over-capacity request was rejected
    let result = timeout(Duration::from_millis(100), response_rx).await;
    assert!(result.is_ok());
    let response = result.unwrap().unwrap();
    assert!(matches!(response, Err(OpRequestError::Failed(msg)) if msg.contains("capacity")));

    // Pending count should still be 5 (rejected request not added)
    assert_eq!(mediator.pending_count(), 5);
}

#[tokio::test]
async fn test_capacity_recovers_after_responses() {
    let (op_tx, op_rx, to_el_tx, to_el_rx, _from_el_tx, from_el_rx) = create_test_channels();
    let mut mediator = TestMediator::new(op_rx, to_el_tx, from_el_rx);
    mediator.max_pending = 3;

    // Fill to capacity
    let mut transactions = Vec::new();
    for _ in 0..3 {
        let tx = Transaction::new::<GetMsg>();
        let (response_tx, _response_rx) = oneshot::channel();
        op_tx.send((tx, response_tx)).await.unwrap();
        transactions.push(tx);
        mediator.process_one_request().await;
    }

    assert_eq!(mediator.pending_count(), 3);

    // Simulate event loop responses for 2 requests
    // In the real mediator, we'd need to create proper OpEnum responses
    // For this test, we verify the pending count decreases

    // Drop event loop receiver to prevent blocking
    drop(to_el_rx);

    // Verify mediator was at capacity
    assert_eq!(mediator.pending_count(), 3);
}

// =============================================================================
// Response Routing Tests
// =============================================================================

#[tokio::test]
async fn test_response_routed_to_correct_executor() {
    // This test verifies that responses are correctly matched to their requests
    let mut pending: std::collections::HashMap<
        Transaction,
        oneshot::Sender<Result<OpEnum, OpRequestError>>,
    > = std::collections::HashMap::new();

    // Create 3 pending requests
    let mut receivers = Vec::new();
    let mut transactions = Vec::new();
    for _ in 0..3 {
        let tx = Transaction::new::<GetMsg>();
        let (response_tx, response_rx) = oneshot::channel();
        pending.insert(tx, response_tx);
        receivers.push(response_rx);
        transactions.push(tx);
    }

    // "Respond" to the middle one
    let target_tx = transactions[1];
    if let Some(sender) = pending.remove(&target_tx) {
        drop(sender.send(Err(OpRequestError::Failed("test response".to_string()))));
    }

    // Only middle receiver should have a response
    for (i, rx) in receivers.into_iter().enumerate() {
        let result = timeout(Duration::from_millis(10), rx).await;
        if i == 1 {
            assert!(result.is_ok(), "Middle request should have response");
        } else {
            assert!(result.is_err(), "Other requests should timeout");
        }
    }
}

#[tokio::test]
async fn test_unknown_response_handled_gracefully() {
    let (op_tx, op_rx, to_el_tx, _to_el_rx, _from_el_tx, from_el_rx) = create_test_channels();
    let mut mediator = TestMediator::new(op_rx, to_el_tx, from_el_rx);

    // Send a response for a transaction that was never registered
    // This simulates a race condition where the executor timed out before the response arrived
    let unknown_tx = Transaction::new::<GetMsg>();

    // We can't easily create a fake OpEnum, but we can verify the mediator
    // handles the case where pending.remove returns None

    // First, register a real request
    let real_tx = Transaction::new::<GetMsg>();
    let (response_tx, _response_rx) = oneshot::channel();
    op_tx.send((real_tx, response_tx)).await.unwrap();
    mediator.process_one_request().await;

    // Verify unknown transaction is not in pending
    assert!(!mediator.pending.contains_key(&unknown_tx));

    // The real implementation logs a warning for unknown transactions
    // but doesn't panic or crash
}

// =============================================================================
// Timeout Behavior Tests
// =============================================================================

#[tokio::test]
async fn test_executor_timeout_handling() {
    let (op_tx, op_rx, to_el_tx, _to_el_rx, _from_el_tx, from_el_rx) = create_test_channels();
    let mut mediator = TestMediator::new(op_rx, to_el_tx, from_el_rx);

    let tx = Transaction::new::<GetMsg>();
    let (response_tx, response_rx) = oneshot::channel();
    op_tx.send((tx, response_tx)).await.unwrap();
    mediator.process_one_request().await;

    // Executor times out waiting for response
    let result = timeout(Duration::from_millis(50), response_rx).await;
    assert!(result.is_err(), "Should timeout waiting for response");

    // Request is still pending in mediator
    assert_eq!(mediator.pending_count(), 1);

    // Mediator doesn't know about timeout until cleanup runs
}

#[tokio::test]
async fn test_concurrent_requests_independent_timeouts() {
    let (op_tx, op_rx, to_el_tx, _to_el_rx, _from_el_tx, from_el_rx) = create_test_channels();
    let mut mediator = TestMediator::new(op_rx, to_el_tx, from_el_rx);

    // Send multiple requests
    let mut receivers = Vec::new();
    for _ in 0..5 {
        let tx = Transaction::new::<GetMsg>();
        let (response_tx, response_rx) = oneshot::channel();
        op_tx.send((tx, response_tx)).await.unwrap();
        receivers.push(response_rx);
        mediator.process_one_request().await;
    }

    // All should timeout independently
    for rx in receivers {
        let result = timeout(Duration::from_millis(10), rx).await;
        assert!(result.is_err(), "Each request should timeout independently");
    }

    // All still pending
    assert_eq!(mediator.pending_count(), 5);
}