Function test_db_handler_concurrent_access

tests/connection_pool_test.rs:74–205

()

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn test_db_handler_concurrent_access() -> Result<(), Box<dyn std::error::Error>> {
    // Use a temporary file for concurrent access
    let temp_dir = tempfile::tempdir()?;
    let db_path = temp_dir.path().join("test_concurrent.db");
    let db_handler = Arc::new(DbHandler::new(db_path.to_str().unwrap())?);
    
    // Create initial session for setup
    let setup_session_id = Uuid::new_v4();
    db_handler.create_session_connection(setup_session_id).await?;
    
    // Create test table with AUTOINCREMENT to avoid conflicts
    db_handler.execute_with_session("CREATE TABLE test_concurrent (id INTEGER PRIMARY KEY AUTOINCREMENT, thread_id INTEGER)", &setup_session_id).await?;
    
    // Test concurrent reads (these should work fine)
    let mut handles = vec![];
    
    // First insert some test data
    for i in 0..10 {
        db_handler.execute_with_session(&format!("INSERT INTO test_concurrent (thread_id) VALUES ({i})"), &setup_session_id)
            .await
            .unwrap_or_else(|e| panic!("Failed to insert initial data {i}: {e:?}"));
    }
    
    // Add a small delay to ensure data is committed (helps with CI timing)
    tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
    
    // Spawn multiple tasks for concurrent reads
    // Use fewer concurrent tasks on CI to reduce resource pressure
    let num_tasks = if std::env::var("CI").is_ok() { 3 } else { 5 };
    let iterations_per_task = if std::env::var("CI").is_ok() { 5 } else { 10 };
    
    for i in 0..num_tasks {
        let db = db_handler.clone();
        let task_iterations = iterations_per_task; // Capture for async block
        let handle = tokio::spawn(async move {
            // Create session for this task
            let task_session_id = Uuid::new_v4();
            db.create_session_connection(task_session_id).await
                .expect("Failed to create session connection");
            
            for j in 0..task_iterations {
                // Retry logic for CI stability
                let mut retry_count = 0;
                let max_retries = if std::env::var("CI").is_ok() { 3 } else { 1 };
                
                loop {
                    // Use COUNT to verify we have data (SELECT has a bug returning only 10 rows)
                    match db.query_with_session("SELECT COUNT(*) FROM test_concurrent", &task_session_id).await {
                        Ok(result) => {
                            // Verify we have at least 10 rows (our initial inserts)
                            if let Some(count_bytes) = &result.rows[0][0] {
                                let count_str = std::str::from_utf8(count_bytes).unwrap();
                                let count: i64 = count_str.parse().unwrap();
                                assert!(count >= 10, "Expected at least 10 rows, got {count}");
                            }
                            break;
                        }
                        Err(e) => {