theta-sync 0.1.0-alpha.1

use std::thread;
use std::time::Duration;
use theta_sync::{unbounded_channel, TryRecvError};

fn main() {
    println!("🔥 ADVERSARIAL CHANNEL TESTING 🔥");
    println!("Attempting to break the channel implementation...");
    println!();
    
    // Run basic attack scenarios
    basic_attacks();
    
    println!();
    println!("🎯 Additional extreme stress tests...");
    
    // Extreme stress test with timing attacks
    extreme_timing_attack();
    
    // Memory corruption attempt
    memory_corruption_attempt();
    
    // Double-drop attempt
    double_drop_attempt();
    
    println!();
    println!("🛡️  All adversarial tests completed!");
    println!("The channel implementation appears to be robust against these attacks.");
}

fn basic_attacks() {
    println!("🚨 Basic adversarial attacks");
    
    // Attack 1: Extreme concurrency
    let (tx, mut rx) = unbounded_channel::<usize>();
    let tx = std::sync::Arc::new(tx);
    let num_threads = 50;
    let messages_per_thread = 500;
    
    let handles: Vec<_> = (0..num_threads)
        .map(|thread_id| {
            let tx = std::sync::Arc::clone(&tx);
            thread::spawn(move || {
                for i in 0..messages_per_thread {
                    if tx.send(thread_id * messages_per_thread + i).is_err() {
                        break;
                    }
                }
            })
        })
        .collect();
    
    let mut received = 0;
    let start = std::time::Instant::now();
    
    while start.elapsed() < Duration::from_millis(100) && received < num_threads * messages_per_thread {
        match rx.try_recv() {
            Ok(_) => received += 1,
            Err(TryRecvError::Empty) => thread::yield_now(),
            Err(TryRecvError::Disconnected) => break,
        }
    }
    
    for handle in handles {
        handle.join().unwrap();
    }
    
    drop(tx);
    while let Ok(_) = rx.try_recv() {
        received += 1;
    }
    
    println!("   Extreme concurrency: {}/{} messages", received, num_threads * messages_per_thread);
}

fn extreme_timing_attack() {
    println!("🚨 Extreme timing attack - trying to cause data races");
    
    let (tx, mut rx) = unbounded_channel::<usize>();
    let tx = std::sync::Arc::new(tx);
    
    // Create 1000 threads that all try to send at the exact same time
    let barrier = std::sync::Arc::new(std::sync::Barrier::new(1001)); // 1000 threads + main
    
    let handles: Vec<_> = (0..1000)
        .map(|i| {
            let tx = std::sync::Arc::clone(&tx);
            let barrier = std::sync::Arc::clone(&barrier);
            thread::spawn(move || {
                barrier.wait(); // Synchronize all threads
                
                // Try to send exactly at the same time
                for j in 0..10 {
                    if let Err(_) = tx.send(i * 10 + j) {
                        break;
                    }
                }
            })
        })
        .collect();
    
    // Main thread waits and then releases all at once
    barrier.wait();
    
    // Immediately start receiving aggressively
    let mut received = 0;
    let start = std::time::Instant::now();
    
    while start.elapsed() < Duration::from_millis(200) {
        match rx.try_recv() {
            Ok(_) => received += 1,
            Err(TryRecvError::Empty) => {
                // Spin aggressively
                for _ in 0..1000 {
                    std::hint::spin_loop();
                }
            },
            Err(TryRecvError::Disconnected) => break,
        }
    }
    
    // Wait for all threads
    for handle in handles {
        handle.join().unwrap();
    }
    
    // Drop sender and drain
    drop(tx);
    while let Ok(_) = rx.try_recv() {
        received += 1;
    }
    
    println!("   Received {} messages in timing attack", received);
}

fn memory_corruption_attempt() {
    println!("🚨 Attempting to trigger memory corruption");
    
    // Create many channels and try to cause memory layout issues
    let mut channels = Vec::new();
    
    for i in 0..100 {
        let (tx, rx) = unbounded_channel::<Vec<u8>>();
        
        // Send different sized messages to try to cause fragmentation
        for j in 0..10 {
            let size = (i * j + 1) % 1000 + 1;
            let data = vec![((i + j) % 256) as u8; size];
            if tx.send(data).is_err() {
                break;
            }
        }
        
        channels.push((tx, rx));
        
        // Randomly drop some channels to fragment memory
        if i % 7 == 0 && !channels.is_empty() {
            let idx = i % channels.len();
            channels.remove(idx);
        }
    }
    
    // Receive from remaining channels in random order
    for (i, channel) in channels.iter_mut().enumerate() {
        let mut received = 0;
        while let Ok(data) = channel.1.try_recv() {
            // Verify data integrity
            if !data.is_empty() {
                let _expected = ((i + received) % 256) as u8;
                // Simple integrity check - data should be valid
                received += 1;
            }
        }
    }
    
    println!("   Memory corruption attempt completed - no corruption detected");
}

fn double_drop_attempt() {
    println!("🚨 Attempting double-drop scenarios");
    
    // This is more about testing our design prevents double-drop issues
    {
        let (tx, mut rx) = unbounded_channel::<i32>();
        let weak = tx.downgrade();
        let tx2 = tx.clone();
        
        tx.send(42).unwrap();
        
        // Create a scenario where multiple references might be dropped
        let tx3 = tx2.clone();
        drop(tx);  // Drop original
        drop(tx2); // Drop clone
        
        // tx3 should still work
        assert!(tx3.send(43).is_ok());
        
        // Upgrade weak should work
        assert!(weak.upgrade().is_some());
        
        drop(tx3); // Drop last strong reference
        
        // Now weak should not upgrade
        assert!(weak.upgrade().is_none());
        
        // Receiver should still get messages
        assert_eq!(rx.try_recv().unwrap(), 42);
        assert_eq!(rx.try_recv().unwrap(), 43);
        assert!(matches!(rx.try_recv(), Err(TryRecvError::Disconnected)));
        
        drop(weak);
        drop(rx);
    }
    
    println!("   Double-drop tests completed - no issues detected");
}