WIP: refactor gateway to use runtime abstraction

README.md

@@ -114,7 +114,7 @@ Our current focus areas include:
 
 - Expanding vector data type capabilities for RAG applications
 - Implementing a test suite to enable end-to-end testing of queries before deployment
-- Building a Deterministic Simulation Testing engine enabling us to robustly iterate faster
+- [Building a Deterministic Simulation Testing engine](docs/simulation-testing.md) enabling us to robustly iterate faster
 - Binary quantisation for even better performance
 
 Long term projects:

docs/simulation-testing.md

@@ -0,0 +1,28 @@
+# Deterministic Simulation Testing
+
+This document outlines the design for a deterministic simulation testing engine inspired by TigerBeetle's VOPR. The goal is to run HelixDB in a fully controlled environment where network, storage and asynchronous execution are scheduled deterministically.
+
+## Motivation
+Deterministic simulation allows us to reproduce race conditions, inject faults and run large workloads at accelerated speed. By running real code under a custom scheduler we can explore many failure scenarios in CI.
+
+## Architecture Overview
+1. **Async Runtime Abstraction**
+   - Introduce an `AsyncRuntime` trait providing primitives such as `spawn` and `sleep`.
+   - Production uses a Tokio backed implementation.
+   - A deterministic scheduler will implement the same trait for testing, controlling the order of task execution.
+
+2. **Transport Abstraction**
+   - Extract a `Transport` trait that wraps accepting and connecting TCP streams.
+   - `ConnectionHandler` and the thread pool depend on this trait instead of `TcpListener`/`TcpStream` directly.
+   - The simulator implements in-memory transport that can delay, drop or reorder packets.
+
+3. **Storage Abstraction**
+   - Define a `Storage` trait around LMDB operations in `helix_engine`.
+   - Implementations include the current LMDB backend and a simulated store capable of injecting errors or latency.
+
+4. **Simulation Harness**
+   - Compose the deterministic runtime, simulated transport and storage.
+   - Run one or more HelixDB instances inside the same process to model a cluster.
+   - Drive workloads, schedule events and collect metrics for fuzzing and invariant checking.
+
+

helix-cli/src/args.rs

@@ -69,13 +69,13 @@ pub enum CommandType {
 #[derive(Debug, Args)]
 #[clap(name = "deploy", about = "Deploy a Helix project")]
 pub struct DeployCommand {
-    #[clap(short, long, help = "The path to the project")]
+    #[clap(short = 'P', long, help = "The path to the project")]
     pub path: Option<String>,
 
     #[clap(short, long, help = "The output path")]
     pub output: Option<String>,
 
-    #[clap(short, long, help = "Port to run the instance on")]
+    #[clap(short = 'p', long, help = "Port to run the instance on")]
     pub port: Option<u16>,
 }
 

helix-container/src/main.rs

@@ -5,6 +5,8 @@ use helixdb::helix_gateway::{
     gateway::{GatewayOpts, HelixGateway},
     router::router::{HandlerFn, HandlerSubmission},
 };
+use helixdb::helix_runtime::tokio_runtime::TokioRuntime;
+use helixdb::helix_transport::tokio_transport::TokioTransport;
 use inventory;
 use std::{collections::HashMap, sync::Arc};
 
@@ -100,11 +102,13 @@ async fn main() {
         GatewayOpts::DEFAULT_POOL_SIZE,
         Some(routes),
         Some(mcp_routes),
+        TokioRuntime::default(),
+        TokioTransport,
     )
     .await;
 
     // start server
     println!("Starting server...");
-    let a = gateway.connection_handler.accept_conns().await.unwrap();
-    let b = a.await.unwrap();
+    let handle = gateway.connection_handler.accept_conns().await.unwrap();
+    handle.await;
 }

helixdb/src/helix_engine/bm25/bm25.rs

@@ -1,4 +1,4 @@
-use heed3::{types::*, Database, Env, RoTxn, RwTxn};
+use crate::helix_storage::heed3::{types::*, Database, Env, RoTxn, RwTxn};
 use serde::{Deserialize, Serialize};
 use std::{borrow::Cow, collections::HashMap, sync::Arc};
 

helixdb/src/helix_engine/bm25/bm25_tests.rs

@@ -5,7 +5,7 @@ mod tests {
         storage_core::storage_core::HelixGraphStorage,
         graph_core::config::Config,
     };
-    use heed3::{EnvOpenOptions, Env};
+    use crate::helix_storage::heed3::{EnvOpenOptions, Env};
     use tempfile::tempdir;
 
     fn setup_test_env() -> (Env, tempfile::TempDir) {
@@ -496,4 +496,4 @@ mod tests {
 
         wtxn.commit().unwrap();
     }
-} 
+} 

helixdb/src/helix_engine/graph_core/ops/bm25/search_bm25.rs

@@ -1,4 +1,4 @@
-use heed3::RoTxn;
+use crate::helix_storage::heed3::RoTxn;
 
 use super::super::tr_val::TraversalVal;
 use crate::helix_engine::{

helixdb/src/helix_engine/graph_core/ops/g.rs

@@ -4,7 +4,7 @@ use crate::helix_engine::{
     storage_core::storage_core::HelixGraphStorage,
     types::GraphError,
 };
-use heed3::{RoTxn, RwTxn};
+use crate::helix_storage::heed3::{RoTxn, RwTxn};
 use std::sync::Arc;
 
 pub struct G {}

helixdb/src/helix_engine/graph_core/ops/in_/in_.rs

@@ -12,15 +12,15 @@ use crate::{
     },
     protocol::label_hash::hash_label,
 };
-use heed3::{types::Bytes, RoTxn};
+use crate::helix_storage::heed3::{types::Bytes, RoTxn};
 use std::sync::Arc;
 
 pub struct InNodesIterator<'a, T> {
-    pub iter: heed3::RoIter<
+    pub iter: crate::helix_storage::heed3::RoIter<
         'a,
         Bytes,
-        heed3::types::LazyDecode<Bytes>,
-        heed3::iteration_method::MoveOnCurrentKeyDuplicates,
+        crate::helix_storage::heed3::types::LazyDecode<Bytes>,
+        crate::helix_storage::heed3::iteration_method::MoveOnCurrentKeyDuplicates,
     >,
     pub storage: Arc<HelixGraphStorage>,
     pub txn: &'a T,

helixdb/src/helix_engine/graph_core/ops/in_/in_e.rs

@@ -9,15 +9,15 @@ use crate::{
     },
     protocol::label_hash::hash_label,
 };
-use heed3::{types::Bytes, RoTxn};
+use crate::helix_storage::heed3::{types::Bytes, RoTxn};
 use std::sync::Arc;
 
 pub struct InEdgesIterator<'a, T> {
-    pub iter: heed3::RoIter<
+    pub iter: crate::helix_storage::heed3::RoIter<
         'a,
         Bytes,
-        heed3::types::LazyDecode<Bytes>,
-        heed3::iteration_method::MoveOnCurrentKeyDuplicates,
+        crate::helix_storage::heed3::types::LazyDecode<Bytes>,
+        crate::helix_storage::heed3::iteration_method::MoveOnCurrentKeyDuplicates,
     >,
     pub storage: Arc<HelixGraphStorage>,
     pub txn: &'a T,

helixdb/src/helix_engine/graph_core/ops/in_/to_n.rs

@@ -3,7 +3,7 @@ use crate::helix_engine::{
     storage_core::{storage_core::HelixGraphStorage, storage_methods::StorageMethods},
     types::GraphError,
 };
-use heed3::RoTxn;
+use crate::helix_storage::heed3::RoTxn;
 use std::sync::Arc;
 
 pub struct ToNIterator<'a, I, T> {

helixdb/src/helix_engine/graph_core/ops/out/from_n.rs

@@ -3,7 +3,7 @@ use crate::helix_engine::{
     storage_core::{storage_core::HelixGraphStorage, storage_methods::StorageMethods},
     types::GraphError,
 };
-use heed3::RoTxn;
+use crate::helix_storage::heed3::RoTxn;
 use std::sync::Arc;
 
 pub struct FromNIterator<'a, I, T> {

helixdb/src/helix_engine/graph_core/ops/out/out.rs

@@ -12,15 +12,15 @@ use crate::{
     },
     protocol::label_hash::hash_label,
 };
-use heed3::{types::Bytes, RoTxn, WithTls};
+use crate::helix_storage::heed3::{types::Bytes, RoTxn, WithTls};
 use std::sync::Arc;
 
 pub struct OutNodesIterator<'a, T> {
-    pub iter: heed3::RoIter<
+    pub iter: crate::helix_storage::heed3::RoIter<
         'a,
         Bytes,
-        heed3::types::LazyDecode<Bytes>,
-        heed3::iteration_method::MoveOnCurrentKeyDuplicates,
+        crate::helix_storage::heed3::types::LazyDecode<Bytes>,
+        crate::helix_storage::heed3::iteration_method::MoveOnCurrentKeyDuplicates,
     >,
     pub storage: Arc<HelixGraphStorage>,
     pub edge_type: &'a EdgeType,

helixdb/src/helix_engine/graph_core/ops/out/out_e.rs

@@ -9,15 +9,15 @@ use crate::{
     },
     protocol::label_hash::hash_label,
 };
-use heed3::{types::Bytes, RoTxn};
+use crate::helix_storage::heed3::{types::Bytes, RoTxn};
 use std::sync::Arc;
 
 pub struct OutEdgesIterator<'a, T> {
-    pub iter: heed3::RoIter<
+    pub iter: crate::helix_storage::heed3::RoIter<
         'a,
         Bytes,
-        heed3::types::LazyDecode<Bytes>,
-        heed3::iteration_method::MoveOnCurrentKeyDuplicates,
+        crate::helix_storage::heed3::types::LazyDecode<Bytes>,
+        crate::helix_storage::heed3::iteration_method::MoveOnCurrentKeyDuplicates,
     >,
     pub storage: Arc<HelixGraphStorage>,
     pub txn: &'a T,

helixdb/src/helix_engine/graph_core/ops/source/add_e.rs

@@ -12,7 +12,7 @@ use crate::{
         value::Value,
     },
 };
-use heed3::PutFlags;
+use crate::helix_storage::heed3::PutFlags;
 use serde::{Deserialize, Serialize};
 
 #[derive(Serialize, Deserialize, Debug, PartialEq, Eq)]

helixdb/src/helix_engine/graph_core/ops/source/add_n.rs

@@ -11,7 +11,7 @@ use crate::{
         value::Value,
     },
 };
-use heed3::PutFlags;
+use crate::helix_storage::heed3::PutFlags;
 
 pub struct AddNIterator {
     inner: std::iter::Once<Result<TraversalVal, GraphError>>,

helixdb/src/helix_engine/graph_core/ops/source/bulk_add_e.rs

@@ -6,7 +6,7 @@ use crate::{
     },
     protocol::{items::Edge, label_hash::hash_label},
 };
-use heed3::PutFlags;
+use crate::helix_storage::heed3::PutFlags;
 
 pub struct BulkAddE {
     inner: std::iter::Once<Result<TraversalVal, GraphError>>,

helixdb/src/helix_engine/graph_core/ops/source/bulk_add_n.rs

@@ -2,7 +2,7 @@ use crate::{
     helix_engine::{graph_core::traversal_iter::RwTraversalIterator, types::GraphError},
     protocol::items::Node,
 };
-use heed3::PutFlags;
+use crate::helix_storage::heed3::PutFlags;
 use super::super::tr_val::TraversalVal;
 
 pub struct BulkAddN {

helixdb/src/helix_engine/graph_core/ops/source/e_from_id.rs

@@ -6,7 +6,7 @@ use crate::{
     },
     protocol::items::Edge,
 };
-use heed3::RoTxn;
+use crate::helix_storage::heed3::RoTxn;
 use std::{iter::Once, sync::Arc};
 
 pub struct EFromId<'a, T> {

helixdb/src/helix_engine/graph_core/ops/source/e_from_type.rs

@@ -5,13 +5,13 @@ use crate::{
     },
     protocol::items::Edge,
 };
-use heed3::{
+use crate::helix_storage::heed3::{
     byteorder::BE,
     types::{Bytes, U128},
 };
 
 pub struct EFromType<'a> {
-    pub iter: heed3::RoIter<'a, U128<BE>, heed3::types::LazyDecode<Bytes>>,
+    pub iter: crate::helix_storage::heed3::RoIter<'a, U128<BE>, crate::helix_storage::heed3::types::LazyDecode<Bytes>>,
     pub label: &'a str,
 }
 

helixdb/src/helix_engine/graph_core/ops/source/n_from_id.rs

@@ -6,7 +6,7 @@ use crate::{
     },
     protocol::items::Node,
 };
-use heed3::RoTxn;
+use crate::helix_storage::heed3::RoTxn;
 use std::{iter::Once, sync::Arc};
 
 pub struct NFromId<'a, T> {

helixdb/src/helix_engine/graph_core/ops/source/n_from_index.rs

@@ -6,13 +6,13 @@ use crate::{
     },
     protocol::{items::Node, value::Value},
 };
-use heed3::{byteorder::BE, types::Bytes, RoTxn};
+use crate::helix_storage::heed3::{byteorder::BE, types::Bytes, RoTxn};
 use serde::Serialize;
 use std::{iter::Once, sync::Arc};
 
 pub struct NFromIndex<'a> {
     iter:
-        heed3::RoPrefix<'a, heed3::types::Bytes, heed3::types::LazyDecode<heed3::types::U128<BE>>>,
+        crate::helix_storage::heed3::RoPrefix<'a, crate::helix_storage::heed3::types::Bytes, crate::helix_storage::heed3::types::LazyDecode<crate::helix_storage::heed3::types::U128<BE>>>,
     txn: &'a RoTxn<'a>,
     storage: Arc<HelixGraphStorage>,
 }

helixdb/src/helix_engine/graph_core/ops/source/n_from_type.rs

@@ -5,13 +5,13 @@ use crate::{
     },
     protocol::items::Node,
 };
-use heed3::{
+use crate::helix_storage::heed3::{
     byteorder::BE,
     types::{Bytes, U128},
 };
 
 pub struct NFromType<'a> {
-    pub iter: heed3::RoIter<'a, U128<BE>, heed3::types::LazyDecode<Bytes>>,
+    pub iter: crate::helix_storage::heed3::RoIter<'a, U128<BE>, crate::helix_storage::heed3::types::LazyDecode<Bytes>>,
     pub label: &'a str,
 }
 

helixdb/src/helix_engine/graph_core/ops/util/drop.rs

@@ -3,7 +3,7 @@ use crate::helix_engine::{
     storage_core::{storage_core::HelixGraphStorage, storage_methods::StorageMethods},
     types::GraphError,
 };
-use heed3::RwTxn;
+use crate::helix_storage::heed3::RwTxn;
 use std::sync::Arc;
 
 pub struct Drop<I> {

helixdb/src/helix_engine/graph_core/ops/util/filter_mut.rs

@@ -1,4 +1,4 @@
-use heed3::RwTxn;
+use crate::helix_storage::heed3::RwTxn;
 
 use crate::helix_engine::{
     graph_core::ops::tr_val::TraversalVal,
@@ -27,4 +27,4 @@ where
             None => None,
         }
     }
-}
+}

helixdb/src/helix_engine/graph_core/ops/util/filter_ref.rs

@@ -1,7 +1,7 @@
 use crate::helix_engine::{graph_core::traversal_iter::RoTraversalIterator, types::GraphError};
 
 use super::super::tr_val::TraversalVal;
-use heed3::RoTxn;
+use crate::helix_storage::heed3::RoTxn;
 
 pub struct FilterRef<'a, I, F> {
     iter: I,

helixdb/src/helix_engine/graph_core/ops/util/map.rs

@@ -4,7 +4,7 @@ use crate::helix_engine::{
 };
 
 use super::super::tr_val::TraversalVal;
-use heed3::RoTxn;
+use crate::helix_storage::heed3::RoTxn;
 
 pub struct Map<'a, I, F> {
     iter: I,

helixdb/src/helix_engine/graph_core/ops/util/paths.rs

@@ -6,7 +6,7 @@ use crate::{
     },
     protocol::{items::Edge, label_hash::hash_label},
 };
-use heed3::RoTxn;
+use crate::helix_storage::heed3::RoTxn;
 use std::{
     collections::{HashMap, HashSet, VecDeque},
     sync::Arc,

helixdb/src/helix_engine/graph_core/ops/vectors/brute_force_search.rs

@@ -1,4 +1,4 @@
-use heed3::RoTxn;
+use crate::helix_storage::heed3::RoTxn;
 
 use super::super::tr_val::TraversalVal;
 use crate::helix_engine::{

helixdb/src/helix_engine/graph_core/ops/vectors/insert.rs

@@ -1,4 +1,4 @@
-use heed3::RoTxn;
+use crate::helix_storage::heed3::RoTxn;
 
 use super::super::tr_val::TraversalVal;
 use crate::{

helixdb/src/helix_engine/graph_core/ops/vectors/search.rs

@@ -1,4 +1,4 @@
-use heed3::RoTxn;
+use crate::helix_storage::heed3::RoTxn;
 
 use super::super::tr_val::TraversalVal;
 use crate::helix_engine::{

helixdb/src/helix_engine/graph_core/traversal_iter.rs

@@ -1,6 +1,6 @@
 use std::sync::Arc;
 
-use heed3::{RoTxn, RwTxn, WithTls};
+use crate::helix_storage::heed3::{RoTxn, RwTxn, WithTls};
 
 use super::ops::tr_val::TraversalVal;
 use crate::helix_engine::{storage_core::storage_core::HelixGraphStorage, types::GraphError};