linkrs-storage 0.1.0

//! Concurrent Operations Integration Tests
//!
//! Tests that concurrent vertex/edge operations maintain data consistency
//! without requiring external synchronization.

mod common;

use linkrs_core::types::VertexId;
use linkrs_core::Value;
use linkrs_storage::{StorageReader, StorageWriter};
use std::sync::{Arc, Barrier, Mutex};
use std::thread;

/// Test: Multiple threads inserting different vertices concurrently
#[test]
fn test_concurrent_vertex_insertion() {
    let mut storage = common::create_in_memory_storage();
    common::setup_basic_schema(&mut storage);

    let storage = Arc::new(Mutex::new(storage));
    let barrier = Arc::new(Barrier::new(8));
    let mut handles = vec![];

    for thread_id in 0..8 {
        let storage = Arc::clone(&storage);
        let barrier = Arc::clone(&barrier);

        let handle = thread::spawn(move || {
            barrier.wait(); // Synchronize start

            for op in 0..100 {
                let vid = VertexId::try_from_int64((thread_id * 100 + op) as i64 + 1000)
                    .expect("test vertex id");
                let name = if let Some(id) = vid.as_int64() {
                    format!("Person_{}", id)
                } else {
                    format!("Person_{}", op)
                };
                let vertex = common::create_person_vertex(
                    vid.as_int64().unwrap_or(op as i64 + 1000),
                    &name,
                    20,
                );

                let mut st = storage.lock().unwrap();
                st.insert_vertex("test_space", vertex)
                    .expect("concurrent insert failed");
            }
        });
        handles.push(handle);
    }

    for handle in handles {
        handle.join().expect("thread panicked");
    }

    // Verify all vertices were inserted
    let st = storage.lock().unwrap();
    let vertices = st.scan_vertices("test_space").unwrap();
    assert_eq!(
        vertices.len(),
        800,
        "Expected 800 vertices from 8 threads * 100 ops"
    );
}

/// Test: Multiple threads reading while another thread modifies
#[test]
fn test_concurrent_read_write_consistency() {
    let mut storage = common::create_in_memory_storage();
    common::setup_basic_schema(&mut storage);

    // Insert initial data
    for i in 0..10 {
        let v = common::create_person_vertex(i, &format!("Person{}", i), 20);
        storage.insert_vertex("test_space", v).unwrap();
    }

    let storage = Arc::new(Mutex::new(storage));
    let barrier = Arc::new(Barrier::new(5));
    let mut handles = vec![];

    // 4 reader threads, 1 writer thread
    for thread_id in 0..5 {
        let storage = Arc::clone(&storage);
        let barrier = Arc::clone(&barrier);

        let handle = thread::spawn(move || {
            barrier.wait();

            if thread_id == 4 {
                // Writer: insert new vertices
                for i in 10..15 {
                    let v = common::create_person_vertex(i, &format!("Person{}", i), 25);
                    let mut st = storage.lock().unwrap();
                    st.insert_vertex("test_space", v).unwrap();
                }
            } else {
                // Readers: scan and verify
                for _ in 0..50 {
                    let st = storage.lock().unwrap();
                    let vertices = st.scan_vertices("test_space").unwrap();
                    assert!(vertices.len() >= 10, "Should have at least 10 vertices");
                }
            }
        });
        handles.push(handle);
    }

    for handle in handles {
        handle.join().expect("thread panicked");
    }
}

/// Test: Concurrent edge creation to same vertex from multiple threads
#[test]
fn test_concurrent_edge_creation_same_target() {
    let mut storage = common::create_in_memory_storage();
    common::setup_basic_schema(&mut storage);

    // Create source and target vertices BEFORE starting concurrent operations
    for i in 0..20 {
        let v = common::create_person_vertex(i, &format!("Person{}", i), 20);
        storage.insert_vertex("test_space", v).unwrap();
    }

    let storage = Arc::new(Mutex::new(storage));
    let barrier = Arc::new(Barrier::new(8));
    let mut handles = vec![];

    // 8 threads creating edges to vertex 1 from different sources
    for thread_id in 0..8 {
        let storage = Arc::clone(&storage);
        let barrier = Arc::clone(&barrier);

        let handle = thread::spawn(move || {
            barrier.wait();

            for edge_idx in 0..10 {
                let src_vid = (thread_id * 10 + edge_idx) as i64 + 2;
                let dst_vid = 1i64;

                let edge = common::create_knows_edge(src_vid, dst_vid, 2020 + edge_idx);

                let mut st = storage.lock().unwrap();
                let result = st.insert_edge("test_space", edge);
                // Some edges might fail if both vertices need to exist in order,
                // but that's OK - we're testing concurrent edge operations
                let _ = result;
            }
        });
        handles.push(handle);
    }

    for handle in handles {
        handle.join().expect("thread panicked");
    }

    // Verify data consistency
    let st = storage.lock().unwrap();
    let all_vertices = st.scan_vertices("test_space").unwrap();
    assert!(
        !all_vertices.is_empty(),
        "Should have vertices after concurrent operations"
    );
}

/// Test: Concurrent index lookup while data is being inserted
#[test]
fn test_concurrent_index_lookup_during_insert() {
    let mut storage = common::create_in_memory_storage();
    common::setup_basic_schema(&mut storage);

    // Create index
    common::create_person_name_index(&mut storage, "test_space");

    let storage = Arc::new(Mutex::new(storage));
    let barrier = Arc::new(Barrier::new(2));
    let mut handles = vec![];

    // Thread 1: Insert vertices
    {
        let storage = Arc::clone(&storage);
        let barrier = Arc::clone(&barrier);

        let handle = thread::spawn(move || {
            barrier.wait();

            for i in 0..50 {
                let v = common::create_person_vertex(i, &format!("Person{}", i), 20);
                let mut st = storage.lock().unwrap();
                st.insert_vertex("test_space", v).unwrap();
            }
        });
        handles.push(handle);
    }

    // Thread 2: Query index
    {
        let storage = Arc::clone(&storage);
        let barrier = Arc::clone(&barrier);

        let handle = thread::spawn(move || {
            barrier.wait();

            for i in 0..25 {
                let st = storage.lock().unwrap();
                let result = st.lookup_index(
                    "test_space",
                    "person_name_idx",
                    &Value::string(format!("Person{}", i)),
                );
                // May or may not find the value depending on timing
                let _ = result;
            }
        });
        handles.push(handle);
    }

    for handle in handles {
        handle.join().expect("thread panicked");
    }
}

/// Test: Concurrent updates don't create inconsistent states
#[test]
fn test_concurrent_vertex_updates_consistency() {
    let mut storage = common::create_in_memory_storage();
    common::setup_basic_schema(&mut storage);

    // Create vertices
    for i in 0..10 {
        let v = common::create_person_vertex(i, &format!("Person{}", i), 20);
        storage.insert_vertex("test_space", v).unwrap();
    }

    let storage = Arc::new(Mutex::new(storage));
    let barrier = Arc::new(Barrier::new(4));
    let mut handles = vec![];

    // 4 threads each reading and verifying vertices
    for thread_id in 0..4 {
        let storage = Arc::clone(&storage);
        let barrier = Arc::clone(&barrier);

        let handle = thread::spawn(move || {
            barrier.wait();

            // Each thread reads different vertices
            for vertex_idx in (thread_id * 2)..(thread_id * 2 + 2) {
                if vertex_idx < 10 {
                    let vid = VertexId::try_from_int64(vertex_idx as i64).expect("test vertex id");

                    let st = storage.lock().unwrap();
                    if let Some(v) = st.get_vertex("test_space", "Person", &vid).unwrap() {
                        // Verify data is consistent
                        assert!(
                            !v.tag.properties.is_empty(),
                            "Vertex should have properties"
                        );
                        assert!(
                            v.properties().contains_key("name"),
                            "Vertex should have name property"
                        );
                    }
                }
            }
        });
        handles.push(handle);
    }

    for handle in handles {
        handle.join().expect("thread panicked");
    }

    // Verify final state
    let st = storage.lock().unwrap();
    let vertices = st.scan_vertices("test_space").unwrap();
    assert_eq!(vertices.len(), 10);
}