Merge pull request #153 from AdaWorldAPI/claude/distributed-db-xor-s3-Ep2Zi

Claude/distributed db xor s3 ep2 zi

Author: AdaWorldAPI

Cargo.toml

@@ -24,7 +24,7 @@ crate-type = ["cdylib", "rlib"]
 # =============================================================================
 
 [features]
-default = ["simd", "parallel"]
+default = ["simd", "parallel", "lancedb", "rustynum"]
 
 # Core features
 simd = []                        # AVX-512 SIMD for Hamming operations
@@ -150,6 +150,7 @@ num-traits = "0.2"
 num_cpus = "1.17"
 itertools = "0.13"
 once_cell = "1.21"
+ctrlc = "3.4"
 lazy_static = "1.5"
 tracing = "0.1"
 tracing-subscriber = "0.3"

crates/ladybug-contract/src/record.rs

@@ -1,33 +1,33 @@
-//! CogRecord — fixed 2 KB record: metadata container + content container.
+//! CogRecord — fixed 4 KB record: metadata container + content container.
 //!
-//! The **holy grail** layout: every record is exactly `[Container; 2]` = 2 KB.
-//! Container 0 is always metadata. Container 1 is always content.
+//! The **holy grail** layout: every record is exactly `[Container; 2]` = 4 KB.
+//! Container 0 is always metadata (16K bits). Container 1 is always content (16K bits).
 //!
 //! ```text
 //! ┌─────────────────────────────────────────────────────────────┐
-//! │  meta    (1 KB)  identity, NARS truth, edges, rung, RL     │
+//! │  meta    (2 KB)  identity, NARS truth, edges, rung, RL     │
 //! ├─────────────────────────────────────────────────────────────┤
-//! │  content (1 KB)  searchable fingerprint (Hamming / SIMD)   │
+//! │  content (2 KB)  searchable fingerprint (Hamming / SIMD)   │
 //! └─────────────────────────────────────────────────────────────┘
-//!       = 2 KB = 1 Fingerprint = 1 DN tree node = 1 Redis value
+//!       = 4 KB = 1 Fingerprint = 1 DN tree node = 1 Redis value
 //! ```
 //!
 //! For multi-container geometries (Xyz, Chunked, Tree), records are
-//! linked through the DN tree. Each linked record is still 2 KB.
+//! linked through the DN tree. Each linked record is still 4 KB.
 
 use std::ops::Range;
 
 use crate::container::Container;
 use crate::geometry::ContainerGeometry;
 use crate::meta::{MetaView, MetaViewMut, W_REPR_BASE};
 
-/// Fixed-size cognitive record: 8,192-bit metadata + 8,192-bit content = 2 KB.
+/// Fixed-size cognitive record: 16,384-bit metadata + 16,384-bit content = 4 KB.
 ///
 /// This is the atomic unit of storage, search, and transfer:
-/// - DN tree value = 1 CogRecord = 2 KB
-/// - Redis value = 1 CogRecord = 2 KB
+/// - DN tree value = 1 CogRecord = 4 KB
+/// - Redis value = 1 CogRecord = 4 KB
 /// - Fingerprint = 1 CogRecord (reinterpretable)
-/// - SIMD scan unit = 2 × 16 AVX-512 iterations = 32 iterations
+/// - SIMD scan unit = 2 × 32 AVX-512 iterations = 64 iterations
 ///
 /// # Stack Allocated
 ///
@@ -43,7 +43,7 @@ pub struct CogRecord {
 }
 
 impl CogRecord {
-    /// Byte size of a CogRecord (2 × 1 KB = 2048 bytes).
+    /// Byte size of a CogRecord (2 × 2 KB = 4096 bytes).
     pub const SIZE: usize = 2 * crate::container::CONTAINER_BYTES;
 
     /// Create a new record with the given geometry.
@@ -108,18 +108,19 @@ impl CogRecord {
         result
     }
 
-    /// Zero-copy byte view of the entire 2 KB record.
+    /// Zero-copy byte view of the entire 4 KB record.
     #[inline]
     pub fn as_bytes(&self) -> &[u8; Self::SIZE] {
         // SAFETY: CogRecord is #[repr(C, align(64))] with two Containers.
-        // Total size = 2 × 1024 = 2048 bytes.
+        // Total size = 2 × 2048 = 4096 bytes.
         unsafe { &*(self as *const CogRecord as *const [u8; Self::SIZE]) }
     }
 
-    /// Construct from a 2 KB byte slice.
+    /// Construct from byte slice (2 × CONTAINER_BYTES).
     pub fn from_bytes(bytes: &[u8; Self::SIZE]) -> Self {
-        let meta_bytes: &[u8; 1024] = bytes[..1024].try_into().unwrap();
-        let content_bytes: &[u8; 1024] = bytes[1024..].try_into().unwrap();
+        use crate::container::CONTAINER_BYTES;
+        let meta_bytes: &[u8; CONTAINER_BYTES] = bytes[..CONTAINER_BYTES].try_into().unwrap();
+        let content_bytes: &[u8; CONTAINER_BYTES] = bytes[CONTAINER_BYTES..].try_into().unwrap();
         Self {
             meta: Container::from_bytes(meta_bytes),
             content: Container::from_bytes(content_bytes),

crates/ladybug-contract/src/wide_meta.rs

@@ -216,11 +216,10 @@ impl<'a> WideMetaView<'a> {
         Self { words }
     }
 
-    /// Get a legacy MetaView over the lower 128 words.
+    /// Get a MetaView over the words (now same width as WideMetaView).
     pub fn legacy(&self) -> crate::meta::MetaView<'_> {
-        // SAFETY: lower 128 words of a 256-word array can be reinterpreted
-        let legacy_words: &[u64; 128] = self.words[..128].try_into().unwrap();
-        crate::meta::MetaView::new(legacy_words)
+        // Container and WideContainer are now the same width (256 words)
+        crate::meta::MetaView::new(self.words)
     }
 
     // ====================================================================
@@ -481,10 +480,9 @@ impl<'a> WideMetaViewMut<'a> {
         WideMetaView { words: self.words }
     }
 
-    /// Get a legacy MetaViewMut over the lower 128 words.
+    /// Get a MetaViewMut over the words (now same width as WideMetaViewMut).
     pub fn legacy_mut(&mut self) -> crate::meta::MetaViewMut<'_> {
-        let legacy_words: &mut [u64; 128] = (&mut self.words[..128]).try_into().unwrap();
-        crate::meta::MetaViewMut::new(legacy_words)
+        crate::meta::MetaViewMut::new(self.words)
     }
 
     // ====================================================================

src/bin/server.rs

@@ -52,6 +52,8 @@ use ladybug::core::simd::{self, hamming_distance};
 use ladybug::nars::TruthValue;
 use ladybug::storage::service::{CognitiveService, CpuFeatures, ServiceConfig};
 use ladybug::storage::{Addr, BindSpace, CogRedis, FINGERPRINT_WORDS, RedisResult};
+#[cfg(feature = "lancedb")]
+use ladybug::storage::LancePersistence;
 use ladybug::{FINGERPRINT_BITS, FINGERPRINT_BYTES, VERSION};
 
 // =============================================================================
@@ -482,6 +484,9 @@ struct DbState {
     cpu: CpuFeatures,
     /// Start time
     start_time: Instant,
+    /// Lance persistence layer
+    #[cfg(feature = "lancedb")]
+    persistence: Arc<LancePersistence>,
 }
 
 impl DbState {
@@ -492,14 +497,71 @@ impl DbState {
         };
 
         let service = CognitiveService::new(svc_config).expect("Failed to create CognitiveService");
+        let mut cog_redis = CogRedis::new();
+        let mut fingerprints = Vec::new();
+
+        #[cfg(feature = "lancedb")]
+        let persistence = {
+            let lance_dir = format!("{}/lance", config.data_dir);
+            let persistence = Arc::new(LancePersistence::new(&lance_dir));
+            let rt = tokio::runtime::Runtime::new().expect("Failed to create tokio runtime");
+
+            if persistence.has_data() {
+                eprintln!("[server] Lance data found, hydrating...");
+
+                if let Err(e) = rt.block_on(persistence.ensure_tables()) {
+                    eprintln!("[server] WARNING: ensure_tables failed: {}", e);
+                }
+
+                match rt.block_on(persistence.hydrate()) {
+                    Ok(Some(space)) => {
+                        let node_count = space.nodes_iter()
+                            .filter(|(a, _)| a.prefix() > 0x0F)
+                            .count();
+                        let edge_count = space.edges_iter().count();
+                        cog_redis.replace_bind_space(space);
+                        eprintln!(
+                            "[server] Hydrated BindSpace: {} nodes, {} edges",
+                            node_count, edge_count
+                        );
+                    }
+                    Ok(None) => {
+                        eprintln!("[server] No BindSpace data to hydrate");
+                    }
+                    Err(e) => {
+                        eprintln!("[server] WARNING: BindSpace hydration failed: {}", e);
+                    }
+                }
+
+                match rt.block_on(persistence.hydrate_index()) {
+                    Ok(fps) if !fps.is_empty() => {
+                        eprintln!("[server] Hydrated {} index fingerprints", fps.len());
+                        fingerprints = fps;
+                    }
+                    Ok(_) => {}
+                    Err(e) => {
+                        eprintln!("[server] WARNING: Index hydration failed: {}", e);
+                    }
+                }
+            } else {
+                eprintln!("[server] No Lance data found, starting fresh");
+                if let Err(e) = rt.block_on(persistence.ensure_tables()) {
+                    eprintln!("[server] WARNING: ensure_tables failed: {}", e);
+                }
+            }
+
+            persistence
+        };
 
         Self {
-            fingerprints: Vec::new(),
+            fingerprints,
             kv: HashMap::new(),
-            cog_redis: CogRedis::new(),
+            cog_redis,
             service,
             cpu: CpuFeatures::detect(),
             start_time: Instant::now(),
+            #[cfg(feature = "lancedb")]
+            persistence,
         }
     }
 }
@@ -1062,7 +1124,7 @@ fn handle_xor(body: &str, format: ResponseFormat) -> Vec<u8> {
     // XOR: one pass over 256 u64 words — nanoseconds
     let mut result = Fingerprint::zero();
     for i in 0..FINGERPRINT_WORDS {
-        result.words_mut()[i] = fp_a.words()[i] ^ fp_b.words()[i];
+        result.as_raw_mut()[i] = fp_a.as_raw()[i] ^ fp_b.as_raw()[i];
     }
 
     match format {
@@ -1105,7 +1167,7 @@ fn handle_xor_verify(body: &str, format: ResponseFormat) -> Vec<u8> {
     // XOR: identical fingerprints produce all zeros
     let mut check = Fingerprint::zero();
     for i in 0..FINGERPRINT_WORDS {
-        check.words_mut()[i] = fp_a.words()[i] ^ fp_b.words()[i];
+        check.as_raw_mut()[i] = fp_a.as_raw()[i] ^ fp_b.as_raw()[i];
     }
     let residual = check.popcount();
     let valid = residual == 0;
@@ -3555,6 +3617,79 @@ fn main() {
 
     let state: SharedState = Arc::new(RwLock::new(DbState::new(&config)));
 
+    // Spawn background persistence thread (flush every 30 seconds)
+    #[cfg(feature = "lancedb")]
+    {
+        let persist_state = Arc::clone(&state);
+        std::thread::spawn(move || {
+            let rt = tokio::runtime::Runtime::new().expect("persist runtime");
+            loop {
+                std::thread::sleep(std::time::Duration::from_secs(30));
+
+                // Check if there are dirty addresses (quick read lock)
+                let has_dirty = {
+                    let db = persist_state.read().unwrap();
+                    db.cog_redis.bind_space().dirty_addrs().next().is_some()
+                };
+
+                if has_dirty {
+                    // Persist under read lock
+                    let persist_result = {
+                        let db = persist_state.read().unwrap();
+                        let persistence = Arc::clone(&db.persistence);
+                        let bs = db.cog_redis.bind_space();
+                        rt.block_on(persistence.persist_full(bs))
+                    };
+
+                    match persist_result {
+                        Ok(()) => {
+                            if let Ok(mut db) = persist_state.write() {
+                                db.cog_redis.bind_space_mut().clear_dirty();
+                            }
+                            eprintln!("[persist] BindSpace flushed to Lance");
+                        }
+                        Err(e) => {
+                            eprintln!("[persist] BindSpace flush failed: {}", e);
+                        }
+                    }
+                }
+
+                // Persist index fingerprints
+                {
+                    let db = persist_state.read().unwrap();
+                    if !db.fingerprints.is_empty() {
+                        let persistence = Arc::clone(&db.persistence);
+                        if let Err(e) = rt.block_on(persistence.persist_index(&db.fingerprints)) {
+                            eprintln!("[persist] Index flush failed: {}", e);
+                        }
+                    }
+                }
+            }
+        });
+    }
+
+    // Register SIGTERM/SIGINT handler for graceful shutdown
+    #[cfg(feature = "lancedb")]
+    {
+        let shutdown_state = Arc::clone(&state);
+        ctrlc::set_handler(move || {
+            eprintln!("\n[shutdown] Flushing to Lance before exit...");
+            let rt = tokio::runtime::Runtime::new().expect("shutdown runtime");
+            let db = shutdown_state.read().unwrap();
+            let persistence = Arc::clone(&db.persistence);
+
+            if let Err(e) = rt.block_on(persistence.persist_full(db.cog_redis.bind_space())) {
+                eprintln!("[shutdown] BindSpace flush failed: {}", e);
+            }
+            if let Err(e) = rt.block_on(persistence.persist_index(&db.fingerprints)) {
+                eprintln!("[shutdown] Index flush failed: {}", e);
+            }
+            eprintln!("[shutdown] Done. Exiting.");
+            std::process::exit(0);
+        })
+        .expect("Failed to set Ctrl-C handler");
+    }
+
     // Spawn UDP bitpacked Hamming listener
     spawn_udp_listener(&config.host, udp_port, Arc::clone(&state));
 
@@ -3565,6 +3700,7 @@ fn main() {
 
     println!("Listening on http://{}", addr);
     println!("UDP Hamming on udp://{}:{}", config.host, udp_port);
+    println!("Lance persistence: {}/lance", config.data_dir);
 
     // Accept connections
     for stream in listener.incoming() {

src/qualia/felt_parse.rs

@@ -888,9 +888,10 @@ impl MirrorField {
         let self_fp = encode_axes(self_axes);
         let partner_fp = encode_axes(partner_axes);
 
-        let mut self_words = [0u64; 128];
-        let mut thou_words = [0u64; 128];
-        for i in 0..128.min(self_fp.len()) {
+        use crate::container::CONTAINER_WORDS;
+        let mut self_words = [0u64; CONTAINER_WORDS];
+        let mut thou_words = [0u64; CONTAINER_WORDS];
+        for i in 0..CONTAINER_WORDS.min(self_fp.len()) {
             self_words[i] = self_fp[i];
             thou_words[i] = partner_fp[i];
         }

src/query/datafusion.rs

@@ -31,7 +31,7 @@ use datafusion::execution::context::SessionContext;
 use datafusion::prelude::*;
 use std::sync::Arc;
 
-use crate::Result;
+use crate::{Error, Result};
 use crate::core::DIM;
 
 // =============================================================================
@@ -124,7 +124,7 @@ impl SqlEngine {
         if std::path::Path::new(&nodes_path).exists() {
             let dataset = Dataset::open(&nodes_path)
                 .await
-                .map_err(|e| Error::Storage(e.to_string()))?;
+                .map_err(|e: lance::Error| Error::Storage(e.to_string()))?;
             let lance_schema = dataset.schema().clone();
             let arrow_schema: ArrowSchema = ArrowSchema::from(&lance_schema);
 
@@ -133,13 +133,13 @@ impl SqlEngine {
                 .scan()
                 .try_into_stream()
                 .await
-                .map_err(|e| Error::Storage(e.to_string()))?;
+                .map_err(|e: lance::Error| Error::Storage(e.to_string()))?;
 
             use futures::StreamExt;
-            let mut all_batches = Vec::new();
+            let mut all_batches: Vec<RecordBatch> = Vec::new();
             let mut stream = batches;
             while let Some(batch) = stream.next().await {
-                all_batches.push(batch.map_err(|e| Error::Storage(e.to_string()))?);
+                all_batches.push(batch.map_err(|e: lance::Error| Error::Storage(e.to_string()))?);
             }
 
             if !all_batches.is_empty() {
@@ -153,21 +153,21 @@ impl SqlEngine {
         if std::path::Path::new(&edges_path).exists() {
             let dataset = Dataset::open(&edges_path)
                 .await
-                .map_err(|e| Error::Storage(e.to_string()))?;
+                .map_err(|e: lance::Error| Error::Storage(e.to_string()))?;
             let lance_schema = dataset.schema().clone();
             let arrow_schema: ArrowSchema = ArrowSchema::from(&lance_schema);
 
             let batches = dataset
                 .scan()
                 .try_into_stream()
                 .await
-                .map_err(|e| Error::Storage(e.to_string()))?;
+                .map_err(|e: lance::Error| Error::Storage(e.to_string()))?;
 
             use futures::StreamExt;
-            let mut all_batches = Vec::new();
+            let mut all_batches: Vec<RecordBatch> = Vec::new();
             let mut stream = batches;
             while let Some(batch) = stream.next().await {
-                all_batches.push(batch.map_err(|e| Error::Storage(e.to_string()))?);
+                all_batches.push(batch.map_err(|e: lance::Error| Error::Storage(e.to_string()))?);
             }
 
             if !all_batches.is_empty() {