Add retry logic with exponential backoff for edge creation

- Added retry_edge_creation helper function to handle transaction conflicts
- Implements exponential backoff with jitter for NotRealizable/TooManyRetry errors
- Up to 5 retries per edge creation attempt
- Fixes double Arc wrapping issue in worker task setup
- Reduces edge creation failures during high-concurrency imports

Author: shisoft

ID_LIST_FIXES_SUMMARY.md

@@ -0,0 +1,114 @@
+# IdList Error Handling Improvements - Summary
+
+## Problem Fixed
+
+The original error messages were vague:
+```
+[ERROR] Cannot find next key, got cell Err(NotRealizable)
+[ERROR] Last segment cell doesn not existed, id None, root Ok(Id {...})
+```
+
+## Solution
+
+**Changed iterator return types from `Option<T>` to `Result<Option<T>, TxnError>`** to properly propagate transaction errors:
+
+### Files Modified
+
+1. **`src/graph/id_list.rs`** - Core iterator implementations
+   - `IdListSegmentIdIterator::next()` and `collect()`
+   - `IdListSegmentIterator::next()` and `last()`
+   - `IdListIterator::next()`, `count()`, `collect()`, `last()`, `next_seg()`
+   - Added detailed diagnostic logging
+
+2. **`src/graph/mod.rs`** - Updated calling code  
+   - Fixed 2 iterator usages to handle `Result`
+   
+3. **`src/graph/vertex/mod.rs`** - Updated calling code
+   - Fixed 1 iterator usage to handle `Result`
+
+### Key Changes
+
+**Before:**
+```rust
+pub async fn next(&mut self) -> Option<Id> {
+    let cell = self.txn.read_selected(...).await;
+    match &cell {
+        Ok(Some(fields)) => { /* ... */ }
+        _ => {
+            error!("Cannot find next key, got cell {:?}", cell)
+        }
+    }
+    None  // ❌ Silently swallows errors
+}
+```
+
+**After:**
+```rust
+pub async fn next(&mut self) -> Result<Option<Id>, TxnError> {
+    let cell = self.txn.read_selected(...).await;
+    match cell {
+        Ok(Some(fields)) => Ok(Some(id)),
+        Ok(None) => {
+            error!("Segment cell not found at level {}, id {:?}", 
+                   self.level, current_id);
+            Ok(None)
+        }
+        Err(e) => {
+            error!("Transaction error reading segment at level {}, id {:?}: {:?}",
+                   self.level, current_id, e);
+            Err(e)  // ✅ Properly propagates errors
+        }
+    }
+}
+```
+
+## Benefits
+
+1. **Better Error Visibility** - Transaction errors now surface with context
+2. **Easier Debugging** - Logs include segment level, IDs, and error types
+3. **Proper Error Propagation** - Errors bubble up instead of being lost
+4. **Production Ready** - Integration tests already cover IdList functionality
+
+## Testing
+
+The IdList module is already thoroughly tested through the graph API in:
+- `src/tests/graph/mod.rs::relationship()` - Comprehensive edge operations
+- `src/tests/graph/mod.rs::edge_operations()` - Edge creation and retrieval
+- `src/tests/graph/mod.rs::graph_traversal()` - Neighborhood queries
+
+All existing tests continue to pass, confirming the improvements don't break functionality.
+
+## Migration Guide
+
+Code using IdList iterators needs to be updated:
+
+```rust
+// OLD - won't compile anymore
+while let Some(id) = iter.next().await {
+    // process
+}
+
+// NEW - handle Result
+loop {
+    match iter.next().await {
+        Ok(Some(id)) => {
+            // process
+        }
+        Ok(None) => break,
+        Err(e) => return Err(e),
+    }
+}
+
+// OR with ? operator
+while let Some(id) = iter.next().await? {
+    // process
+}
+```
+
+## Status
+
+✅ **All changes compile successfully**  
+✅ **Existing tests pass**  
+✅ **Better error diagnostics in place**
+
+

src/apps/wikidata/importer.rs

@@ -23,6 +23,8 @@ use tokio::task;
 use std::time::{Instant, Duration};
 use std::sync::atomic::{AtomicUsize, Ordering};
 use futures::stream::{FuturesUnordered, StreamExt};
+use neb::client::transaction::TxnError;
+use crate::graph::LinkVerticesError;
 
 #[derive(Error, Debug)]
 pub enum ImportError {
@@ -128,6 +130,56 @@ pub fn create_optimized_line_reader(file_path: &str) -> Result<Box<dyn Iterator<
     }
 }
 
+/// Helper function to retry edge creation with exponential backoff
+/// Handles transaction errors like NotRealizable and TooManyRetry
+async fn retry_edge_creation<F, Fut>(
+    server: Arc<MorpheusServer>,
+    from_id: Id,
+    to_id: Id,
+    edge_data: Option<OwnedMap>,
+    create_fn: F,
+    max_retries: u32,
+) -> Result<(), String>
+where
+    F: Fn(Arc<MorpheusServer>, Id, Id, Option<OwnedMap>) -> Fut,
+    Fut: std::future::Future<Output = Result<Result<crate::graph::edge::Edge, LinkVerticesError>, TxnError>>,
+{
+    let mut retry_count = 0;
+    let base_delay = Duration::from_millis(10);
+    
+    loop {
+        match create_fn(server.clone(), from_id, to_id, edge_data.clone()).await {
+            Ok(Ok(_)) => return Ok(()),
+            Ok(Err(e)) => {
+                // Non-transaction error (e.g., schema not found), don't retry
+                return Err(format!("{:?}", e));
+            }
+            Err(e) => {
+                // Transaction error - check if we should retry
+                let error_str = format!("{:?}", e);
+                let should_retry = error_str.contains("NotRealizable") 
+                    || error_str.contains("TooManyRetry")
+                    || error_str.contains("Conflict")
+                    || error_str.contains("Transaction");
+                
+                if should_retry && retry_count < max_retries {
+                    retry_count += 1;
+                    // Exponential backoff with jitter (using retry_count for pseudo-randomness)
+                    let delay = base_delay * (1 << retry_count.min(8)); // Cap at 2^8 = 256 * base_delay
+                    let jitter = Duration::from_millis((retry_count as u64 * 7) % 50); // Simple pseudo-random jitter
+                    let total_delay = delay + jitter;
+                    
+                    tokio::time::sleep(total_delay).await;
+                    continue;
+                } else {
+                    // Max retries reached or non-retriable error
+                    return Err(format!("Failed after {} retries: {:?}", retry_count, e));
+                }
+            }
+        }
+    }
+}
+
 /// Wikidata importer with three-phase import strategy
 /// Phase 1: Import all entities and properties as vertices
 /// Phase 2: Import statements and create graph edges for Q→Q relationships
@@ -815,7 +867,7 @@ impl WikidataImporter {
 
         // Spawn worker tasks
         let mut worker_handles = Vec::new();
-        let server = Arc::new(self.server.clone());
+        let server = self.server.clone(); // Already an Arc
         let parser = self.parser.clone();
         let entity_id_map = self.entity_id_map.clone();
 
@@ -959,13 +1011,25 @@ impl WikidataImporter {
                                                 literal_id: None,
                                             }.to_owned_map();
 
-                                            match server_clone.graph.link(entity_id, "wikidata_link", target_entity_id, &Some(edge_data)).await {
+                                            // Use retry helper for edge creation with exponential backoff
+                                            match retry_edge_creation(
+                                                server_clone,
+                                                entity_id,
+                                                target_entity_id,
+                                                Some(edge_data),
+                                                |srv, from, to, body| {
+                                                    async move {
+                                                        srv.graph.link(from, "wikidata_link", to, &body).await
+                                                    }
+                                                },
+                                                5, // max 5 retries
+                                            ).await {
                                                 Ok(_) => {
                                                     edges_created_clone.fetch_add(1, Ordering::Relaxed);
                                                     local_stats.1 += 1;
                                                 }
                                                 Err(e) => {
-                                                    eprintln!("Worker {}: Failed to create edge {:?} -> {:?}: {:?}", worker_id, entity_id, target_entity_id, e);
+                                                    eprintln!("Worker {}: Failed to create edge {:?} -> {:?}: {}", worker_id, entity_id, target_entity_id, e);
                                                     local_stats.3 += 1;
                                                 }
                                             }