added local

varshith-Git · varshith-Git · commit cab20b78a8aa · 2025-12-26T13:05:14.000+05:30
diff --git a/README.md b/README.md
diff --git a/node/src/api.rs b/node/src/api.rs
@@ -166,3 +166,14 @@ pub struct EventProofResponse {
     pub event_count: u64,
     pub committed_height: u64,
 }
+
+// Phase 34: Batch Ingestion
+#[derive(Deserialize, Serialize, Debug)]
+pub struct BatchInsertRequest {
+    pub batch: Vec<Vec<f32>>,
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+pub struct BatchInsertResponse {
+    pub ids: Vec<u32>,
+}
diff --git a/node/src/engine.rs b/node/src/engine.rs
@@ -268,6 +268,105 @@ impl<const MAX_RECORDS: usize, const D: usize, const MAX_NODES: usize, const MAX
         }
     }
 
+    /// Insert a batch of records in a single atomic transaction.
+    /// Returns the list of assigned IDs.
+    pub fn insert_batch(&mut self, batch_values: &[Vec<f32>]) -> Result<Vec<u32>, EngineError> {
+        if batch_values.is_empty() {
+            return Ok(Vec::new());
+        }
+
+        // Validate all inputs first
+        for values in batch_values {
+            if values.len() != D {
+                return Err(EngineError::InvalidInput(format!("Expected {} dimensions, got {}", D, values.len())));
+            }
+            for &v in values {
+                if v > MAX_SAFE_F || v < MIN_SAFE_F {
+                    return Err(EngineError::InvalidInput(format!(
+                        "Embedding value {} out of allowed range [{:.1}, {:.1}]",
+                        v, MIN_SAFE_F, MAX_SAFE_F
+                    )));
+                }
+            }
+        }
+
+        // Prepare events
+        // Phase 23: Event-sourced path ONLY (Batching not supported in legacy WAL)
+        if let Some(ref mut committer) = self.event_committer {
+            let mut events = Vec::with_capacity(batch_values.len());
+            let mut assigned_ids = Vec::with_capacity(batch_values.len());
+            
+            // Track used IDs to avoid collisions within the batch
+            let mut temp_used_ids = std::collections::HashSet::new();
+
+            // ID Generation Logic (Provisioning)
+            let mut next_candidate = 0;
+
+            for values in batch_values {
+                // Find next free slot
+                let mut found_id = None;
+                for i in next_candidate..MAX_RECORDS {
+                    let rid = RecordId(i as u32);
+                    // Check if occupied in Kernel OR already assigned in this batch
+                    if self.state.get_record(rid).is_none() && !temp_used_ids.contains(&i) {
+                        found_id = Some(rid);
+                        next_candidate = i + 1; // Optimization: start next search here
+                        break;
+                    }
+                }
+                
+                let id = found_id.ok_or(valori_kernel::error::KernelError::CapacityExceeded)?;
+                temp_used_ids.insert(id.0 as usize);
+                assigned_ids.push(id.0);
+
+                // Create FxpVector
+                let mut vector = FxpVector::<D>::new_zeros();
+                for (i, v) in values.iter().enumerate() {
+                    let fixed = (v * SCALE).round().clamp(i32::MIN as f32, i32::MAX as f32) as i32;
+                    vector.data[i] = FxpScalar(fixed);
+                }
+
+                events.push(KernelEvent::InsertRecord { id, vector });
+            }
+
+            let start = std::time::Instant::now();
+            
+            // Atomic Batch Commit
+            match committer.commit_batch(events.clone()) {
+                Ok(CommitResult::Committed) => {
+                     tracing::info!("Batch committed: {} records", events.len());
+                     metrics::counter!("valori_events_committed_total", events.len() as u64);
+                     metrics::histogram!("valori_batch_commit_duration_seconds", start.elapsed().as_secs_f64());
+
+                     // Sync State & Index
+                     for event in &events {
+                         self.state.apply_event(event).map_err(EngineError::Kernel)?;
+                         
+                         if let KernelEvent::InsertRecord { id, vector } = event {
+                             let mut consistent_values = Vec::with_capacity(D);
+                             for i in 0..D {
+                                 let fxp = vector.data[i];
+                                 let f = fxp.0 as f32 / SCALE;
+                                 consistent_values.push(f);
+                             }
+                             self.index.insert(id.0, &consistent_values);
+                         }
+                     }
+                     
+                     Ok(assigned_ids)
+                },
+                Ok(CommitResult::RolledBack) => {
+                    Err(EngineError::InvalidInput("Batch validation failed (Rolled Back)".to_string()))
+                },
+                Err(e) => {
+                    Err(EngineError::InvalidInput(format!("Batch commit failed: {:?}", e)))
+                }
+            }
+        } else {
+            Err(EngineError::InvalidInput("Batch insert requires Event Log (legacy WAL not supported)".to_string()))
+        }
+    }
+
     /// Apply an event that has already been committed (e.g. from replication stream or local commit).
     /// Updates BOTH kernel state AND auxiliary structures (Index, Bitmap).
     pub fn apply_committed_event(&mut self, event: &KernelEvent<D>) -> Result<(), EngineError> {
diff --git a/node/src/events/event_commit.rs b/node/src/events/event_commit.rs
@@ -233,11 +233,12 @@ impl<const M: usize, const D: usize, const N: usize, const E: usize> EventCommit
             return Ok(CommitResult::Committed);
         }
 
-        // Step 1: Persist ALL events to disk first
-        for event in &events {
-            let entry = crate::events::event_log::LogEntry::Event(event.clone());
-            self.event_log.append(&entry)?;
-        }
+        // Step 1: Persist ALL events to disk first (Single Fsync)
+        let log_entries: Vec<_> = events.iter()
+            .map(|e| crate::events::event_log::LogEntry::Event(e.clone()))
+            .collect();
+            
+        self.event_log.append_batch(&log_entries)?;
 
         // Step 2: Add all to buffer
         for event in &events {
diff --git a/node/src/events/event_log.rs b/node/src/events/event_log.rs
@@ -206,6 +206,40 @@ impl<const D: usize> EventLogWriter<D> {
         Ok(())
     }
 
+    /// Append multiple entries to the log with a SINGLE fsync
+    ///
+    /// This provides atomicity for batches: either all specific bytes are physically on disk
+    /// (after fsync return) or we crash before fsync returns (and they might not be).
+    ///
+    /// Note: If a partial write happens (less than full batch), the log recovery
+    /// logic must handle truncation of incomplete tail writes.
+    pub fn append_batch(&mut self, entries: &[LogEntry<D>]) -> Result<()> {
+        if entries.is_empty() {
+             return Ok(());
+        }
+
+        for entry in entries {
+            let bytes = bincode::serde::encode_to_vec(entry, bincode::config::standard())
+                .map_err(|e| EventLogError::Serialization(e.to_string()))?;
+            self.file.write_all(&bytes)?;
+        }
+        
+        // Flush buffer once
+        self.file.flush()?;
+        
+        // Force fsync once
+        self.file.get_ref().sync_all()?;
+
+        // Update counts
+        for entry in entries {
+            if let LogEntry::Event(_) = entry {
+                self.event_count += 1;
+            }
+        }
+        
+        Ok(())
+    }
+
     /// Get the number of events written
     pub fn event_count(&self) -> u64 {
         self.event_count
diff --git a/node/src/server.rs b/node/src/server.rs
@@ -81,6 +81,7 @@ pub fn build_router<const M: usize, const D: usize, const N: usize, const E: usi
 ) -> Router {
     let mut app = Router::new()
         .route("/records", post(insert_record))
+        .route("/v1/vectors/batch_insert", post(batch_insert)) // Phase 34
         .route("/search", post(search))
         .route("/graph/node", post(create_node))
         .route("/graph/edge", post(create_edge))
@@ -184,6 +185,15 @@ async fn insert_record<const M: usize, const D: usize, const N: usize, const E:
     Ok(Json(InsertRecordResponse { id }))
 }
 
+async fn batch_insert<const M: usize, const D: usize, const N: usize, const E: usize>(
+    State(state): State<SharedEngine<M, D, N, E>>,
+    Json(payload): Json<BatchInsertRequest>,
+) -> Result<Json<BatchInsertResponse>, EngineError> {
+    let mut engine = state.lock().await;
+    let ids = engine.insert_batch(&payload.batch)?;
+    Ok(Json(BatchInsertResponse { ids }))
+}
+
 async fn search<const M: usize, const D: usize, const N: usize, const E: usize>(
     State(state): State<SharedEngine<M, D, N, E>>,
     Json(payload): Json<SearchRequest>,
diff --git a/node/tests/api_batch_ingest.rs b/node/tests/api_batch_ingest.rs
@@ -0,0 +1,108 @@
+use valori_node::config::NodeConfig;
+use valori_node::server::build_router;
+use valori_node::engine::Engine;
+use valori_node::api::{BatchInsertRequest, BatchInsertResponse, InsertRecordRequest};
+use axum::{
+    body::Body,
+    http::{Request, StatusCode},
+};
+use tower::ServiceExt; // for oneshot
+use std::sync::Arc;
+use tokio::sync::Mutex;
+use tempfile::tempdir;
+
+// Define concrete types matching server.rs
+const M: usize = 100;
+const D: usize = 16;
+const N: usize = 100;
+const E: usize = 200;
+
+#[tokio::test]
+async fn test_batch_ingest_success() {
+    let dir = tempdir().unwrap();
+    let db_path = dir.path().join("valori.wal");
+    let event_log_path = dir.path().join("events.log");
+
+    let mut config = NodeConfig::default();
+    config.max_records = M;
+    config.dim = D;
+    config.max_nodes = N;
+    config.max_edges = E;
+    config.wal_path = Some(db_path.clone());
+    config.event_log_path = Some(event_log_path.clone()); // Enable Event Log for Batching
+
+    let engine = Engine::<M, D, N, E>::new(&config);
+    let shared_state = Arc::new(Mutex::new(engine));
+    let app = build_router(shared_state, None);
+
+    // Prepare Batch
+    let batch = vec![
+        vec![0.1; D],
+        vec![0.2; D],
+        vec![0.3; D],
+    ];
+
+    let req = Request::builder()
+        .method("POST")
+        .uri("/v1/vectors/batch_insert")
+        .header("content-type", "application/json")
+        .body(Body::from(serde_json::to_vec(&BatchInsertRequest { batch }).unwrap()))
+        .unwrap();
+
+    let response = app.oneshot(req).await.unwrap();
+    assert_eq!(response.status(), StatusCode::OK);
+
+    let body_bytes = axum::body::to_bytes(response.into_body(), 1024).await.unwrap();
+    let resp: BatchInsertResponse = serde_json::from_slice(&body_bytes).unwrap();
+
+    assert_eq!(resp.ids.len(), 3);
+    assert_eq!(resp.ids, vec![0, 1, 2]); // First batch should get 0, 1, 2
+}
+
+#[tokio::test]
+async fn test_batch_ingest_atomicity_failure() {
+    let dir = tempdir().unwrap();
+    let db_path = dir.path().join("valori.wal");
+    let event_log_path = dir.path().join("events.log");
+
+    let mut config = NodeConfig::default();
+    config.max_records = M;
+    config.dim = D;
+    config.max_nodes = N;
+    config.max_edges = E;
+    config.wal_path = Some(db_path.clone());
+    config.event_log_path = Some(event_log_path.clone());
+
+    let engine = Engine::<M, D, N, E>::new(&config);
+    let shared_state = Arc::new(Mutex::new(engine));
+    let app = build_router(shared_state.clone(), None);
+
+    // Invalid payload (one vector has wrong dim)
+    let batch = vec![
+        vec![0.1; D],
+        vec![0.2; D + 1], // INVALID DIM
+        vec![0.3; D],
+    ];
+
+    let req = Request::builder()
+        .method("POST")
+        .uri("/v1/vectors/batch_insert")
+        .header("content-type", "application/json")
+        .body(Body::from(serde_json::to_vec(&BatchInsertRequest { batch }).unwrap()))
+        .unwrap();
+
+    let response = app.oneshot(req).await.unwrap();
+    // Should fail validation before commit
+    // Since insert_batch validates strictly before commit, this should return 500 or 400 depending on error mapping
+    // EngineError::InvalidInput maps to INTERNAL_SERVER_ERROR currently? or BAD_REQUEST?
+    // Let's check api.rs/errors.rs mapping. Usually InvalidInput -> 400?
+    // Actually, Axum doesn't auto-map EngineError. 
+    // Wait, EngineError needs IntoResponse.
+    // Assuming standard error handling returns error code.
+    assert!(response.status().is_client_error() || response.status().is_server_error());
+
+    // Verify NOTHING was inserted
+    let engine = shared_state.lock().await;
+    // Check ID 0 is empty
+    assert!(engine.search_l2(&vec![0.1; D], 1).unwrap().is_empty());
+}
diff --git a/python/valori/__init__.py b/python/valori/__init__.py
@@ -4,16 +4,17 @@
 from .remote import RemoteClient
 
 class Valori:
-    def __new__(cls, remote: Optional[str] = None):
+    def __new__(cls, remote: Optional[str] = None, path: str = "./valori_db"):
         """
         Factory yielding either a LocalClient (FFI) or RemoteClient (HTTP).
         
         Args:
             remote: If None (default), uses LocalClient (ffi). 
                     If a URL string, uses RemoteClient.
+            path: Path to database directory (only used for LocalClient).
         """
         if remote is None:
-            return LocalClient()
+            return LocalClient(path=path)
         else:
             return RemoteClient(base_url=remote)
 
diff --git a/python/valori/local.py b/python/valori/local.py
@@ -15,10 +15,10 @@
         _ffi = None
 
 class LocalClient:
-    def __init__(self):
+    def __init__(self, path: str = "./valori_db"):
         if _ffi is None:
              raise ImportError("Could not load 'valori_ffi' module. Ensure it is compiled and in PYTHONPATH.")
-        self.kernel = _ffi.PyKernel()
+        self.kernel = _ffi.ValoriEngine(path)
 
     def insert(self, vector: List[float]) -> int:
         return self.kernel.insert(vector)
diff --git a/python/valori/remote.py b/python/valori/remote.py
@@ -19,6 +19,12 @@ def insert(self, vector: List[float]) -> int:
         resp = self._post("/records", data)
         return resp["id"]
 
+    def insert_batch(self, batch: List[List[float]]) -> List[int]:
+        """Insert a batch of vectors. Returns list of new Record IDs."""
+        data = {"batch": batch}
+        resp = self._post("/v1/vectors/batch_insert", data)
+        return resp["ids"]
+
     def search(self, query: List[float], k: int) -> List[Dict[str, Any]]:
         """Search for nearest vectors. Returns list of hits [{'id': int, 'score': int}]."""
         data = {"query": query, "k": k}
