Add the FDB async driver.

- Adds all the machinery to run FDB.
- Update the connection options to be able to select drivers.
- Update code to use the Driver interface depending on driver.
- Update testing harness to be able to read a ServerConfig file.

Author: calavera

Cargo.lock

@@ -7,88 +7,81 @@ license = "Apache-2.0"
 build = "build.rs"
 resolver = "2"
 
-# [lib]
-# name = "indexify_server"
-# path = "src/lib.rs"
-
 [features]
 default = []
 migrations = []
 console-subscriber = ["dep:console-subscriber"]
 
 [dependencies]
-async-stream = "0.3.6"
-serde = { version = "1.0.228", features = ["derive"] }
-serde_json = "1.0.145"
 anyhow = "1.0.100"
-figment = { version = "0.10.19", features = ["yaml"] }
-clap = { version = "4.5.50", features = ["derive"] }
-tracing = "0.1.41"
+async-stream = "0.3.6"
+async-trait = "0.1.89"
 axum = { version = "0.8.6", features = ["multipart", "macros", "tokio"] }
-tokio = { version = "1.48.0", features = ["full"] }
 axum-server = "0.7.2"
-futures = "0.3.31"
-tracing-subscriber = { version = "0.3.20", features = ["env-filter", "json"] }
-utoipa = { version = "5.4.0", features = ["axum_extras", "yaml"] }
-utoipa-swagger-ui = { version = "9.0.2", features = ["axum"] }
-sha2 = "0.10.9"
-nanoid = "0.4.0"
-object_store = { version = "0.12.4", features = ["aws"] }
-uuid = { version = "1.18.1", features = ["v4"] }
-tower-http = { version = "0.6.6", default-features = false, features = ["cors", "trace"] }
+axum-tracing-opentelemetry = { version = "0.32.1", features = ["tracing_level_info"] }
+base64 = "0.22.1"
 bytes = { version = "1.10.1", features = ["serde"] }
+chrono = "0.4.42"
 ciborium = "0.2.2"
-rand = "0.9.2"
+clap = { version = "4.5.50", features = ["derive"] }
+console-subscriber = { version = "0.5.0", optional = true }
+derive_builder = "0.20.2"
+figment = { version = "0.10.19", features = ["yaml"] }
+foundationdb = { version = "0.10.0", features = ["embedded-fdb-include", "fdb-7_4", "num-bigint", "serde", "serde_bytes", "serde_json"] }
+futures = "0.3.31"
 hyper = "1.7.0"
-url = "2.5.6"
+im = { version = "15.1.0" }
+json-subscriber = "0.2.6"
+nanoid = "0.4.0"
+object_store = { version = "0.12.4", features = ["aws"] }
+omniqueue = { git = "https://github.com/tensorlakeai/omniqueue-rs", branch = "main", features = ["sqs"], default-features = false }
 opentelemetry = { version = "0.31.0", features = ["metrics", "trace"] }
-opentelemetry_sdk = { version = "0.31.0", features = ["rt-tokio", "metrics", "trace"] }
 opentelemetry-otlp = { version = "0.31.0", features = ["tokio", "metrics", "trace", "grpc-tonic", "zstd-tonic"] }
-axum-tracing-opentelemetry = { version = "0.32.1", features = ["tracing_level_info"] }
-base64 = "0.22.1"
-tonic = "0.14.2"
-prost = "0.14.1"
-tonic-prost = "0.14.2"
-tonic-reflection = "0.14.2"
-tokio-stream = { version = "0.1.17", features = ["sync"] }
-priority-queue = "2.6.0"
-slatedb = { version = "0.8.2" }
-rust-embed = { version = "8.8.0", features = ["mime-guess"] }
-pin-project-lite = "0.2.16"
+opentelemetry_sdk = { version = "0.31.0", features = ["rt-tokio", "metrics", "trace"] }
+opentelemetry-stdout = { version = "0.31.0", default-features = false, features = ["trace"] }
 pin-project = "1.1.10"
-im = { version = "15.1.0" }
-async-trait = "0.1.89"
+pin-project-lite = "0.2.16"
+priority-queue = "2.6.0"
+prost = "0.14.1"
+prost-types = "0.14.1"
+rand = "0.9.2"
 rocksdb = { version = "0.24.0" }
+rust-embed = { version = "8.8.0", features = ["mime-guess"] }
+serde = { version = "1.0.228", features = ["derive"] }
+serde_json = "1.0.145"
+sha2 = "0.10.9"
+slatedb = { version = "0.8.2" }
 strum = { version = "0.27.2", features = ["derive"] }
-derive_builder = "0.20.2"
 tempfile = "3.23.0"
 thiserror = "2.0.17"
-json-subscriber = "0.2.6"
-omniqueue = { git = "https://github.com/tensorlakeai/omniqueue-rs", branch = "main", features = ["sqs"], default-features = false }
-chrono = "0.4.42"
-prost-types = "0.14.1"
+tokio = { version = "1.48.0", features = ["full"] }
+tokio-stream = { version = "0.1.17", features = ["sync"] }
+tonic = "0.14.2"
+tonic-prost = "0.14.2"
+tonic-reflection = "0.14.2"
+tower-http = { version = "0.6.6", default-features = false, features = ["cors", "trace"] }
+tracing = "0.1.41"
 tracing-opentelemetry = "0.32.0"
-opentelemetry-stdout = { version = "0.31.0", default-features = false, features = ["trace"] }
-console-subscriber = { version = "0.5.0", optional = true }
+tracing-subscriber = { version = "0.3.20", features = ["env-filter", "json"] }
+url = "2.5.6"
+utoipa = { version = "5.4.0", features = ["axum_extras", "yaml"] }
+utoipa-swagger-ui = { version = "9.0.2", features = ["axum"] }
+uuid = { version = "1.18.1", features = ["v4"] }
 
 
 [dev-dependencies]
-tempfile = "3.20.0"
-strum = { version = "0.27.1", features = ["derive"] }
+criterion = { version = "0.5", features = ["html_reports"] }
+mock_instant = "0.6.0"
 rocksdb = { version = "0.24.0" }
+strum = { version = "0.27.1", features = ["derive"] }
+tempfile = "3.20.0"
 tokio = { version = "1.45.1", features = ["full", "test-util"] }
-mock_instant = "0.6.0"
-criterion = { version = "0.5", features = ["html_reports"] }
-
-# [[bench]]
-# name = "state_store"
-# harness = false
 
 [build-dependencies]
 anyhow = "1.0.98"
+tonic-prost-build = "0.14.2"
 # All features enabled
 vergen = { version = "9.0.6", features = ["build", "cargo", "rustc", "si"] }
-tonic-prost-build = "0.14.2"
 
 [package.metadata.deb]
 maintainer = "Diptanu Gon Choudhury <[email protected]>"

server/Cargo.toml

@@ -1,7 +1,7 @@
 use std::{
-    env,
     fmt::{Debug, Display},
     net::SocketAddr,
+    path::Path,
     time::Duration,
 };
 
@@ -13,7 +13,7 @@ use figment::{
 use serde::{Deserialize, Serialize};
 use uuid::Uuid;
 
-use crate::{blob_store::BlobStorageConfig, state_store::driver::rocksdb::RocksDBConfig};
+use crate::{blob_store::BlobStorageConfig, state_store::driver::ConnectionOptions};
 
 const LOCAL_ENV: &str = "local";
 
@@ -59,8 +59,7 @@ impl Display for ExecutorCatalogEntry {
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct ServerConfig {
     pub env: String,
-    pub state_store_path: String,
-    pub rocksdb_config: RocksDBConfig,
+    pub driver_config: ConnectionOptions,
     pub listen_addr: String,
     pub listen_addr_grpc: String,
     pub blob_storage: BlobStorageConfig,
@@ -73,11 +72,9 @@ pub struct ServerConfig {
 
 impl Default for ServerConfig {
     fn default() -> Self {
-        let state_store_path = env::current_dir().unwrap().join("indexify_storage/state");
         ServerConfig {
             env: LOCAL_ENV.to_string(),
-            state_store_path: state_store_path.to_str().unwrap().to_string(),
-            rocksdb_config: RocksDBConfig::default(),
+            driver_config: ConnectionOptions::default(),
             listen_addr: "0.0.0.0:8900".to_string(),
             listen_addr_grpc: "0.0.0.0:8901".to_string(),
             blob_storage: Default::default(),
@@ -91,8 +88,8 @@ impl Default for ServerConfig {
 }
 
 impl ServerConfig {
-    pub fn from_path(path: &str) -> Result<ServerConfig> {
-        let config_str = std::fs::read_to_string(path)?;
+    pub fn from_path<P: AsRef<Path>>(path: P) -> Result<ServerConfig> {
+        let config_str = std::fs::read_to_string(path.as_ref())?;
         let figment = Figment::from(Serialized::defaults(ServerConfig::default()));
 
         let config: ServerConfig = figment.merge(Yaml::string(&config_str)).extract()?;

server/src/config.rs

@@ -15,7 +15,6 @@ mod tests {
         },
         service::Service,
         state_store::{
-            driver::Writer,
             requests::{
                 CreateOrUpdateApplicationRequest,
                 DeleteApplicationRequest,

server/src/gc_test.rs

@@ -27,7 +27,7 @@ mod tests {
         executors::EXECUTOR_TIMEOUT,
         service::Service,
         state_store::{
-            driver::{IterOptions, Reader, rocksdb::RocksDBDriver},
+            driver::{Driver, IterOptions},
             requests::{
                 CreateOrUpdateApplicationRequest,
                 DeleteApplicationRequest,
@@ -44,7 +44,7 @@ mod tests {
     const TEST_FN_MAX_RETRIES: u32 = 3;
 
     async fn assert_cf_counts(
-        db: Arc<RocksDBDriver>,
+        db: Arc<dyn Driver>,
         mut asserts: HashMap<String, usize>,
     ) -> Result<()> {
         if !asserts.contains_key(IndexifyObjectsColumns::StateMachineMetadata.as_ref()) {

server/src/integration_test.rs

@@ -52,7 +52,7 @@ struct Cli {
 async fn main() -> anyhow::Result<()> {
     let cli = Cli::parse();
     let config = match cli.config {
-        Some(path) => config::ServerConfig::from_path(path.to_str().unwrap()).unwrap(),
+        Some(path) => config::ServerConfig::from_path(&path)?,
         None => config::ServerConfig::default(),
     };