feat: Added CI for in_memory_testing

Author: Harsh Dev Pathak

.github/workflows/build-and-test-in-memory.yml

@@ -0,0 +1,70 @@
+name: In-Memory VSS Server CI
+
+on:
+  push:
+    branches: [ main ]
+  pull_request:
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  test-in-memory:
+    runs-on: ubuntu-latest
+    timeout-minutes: 5
+
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v4
+
+      - name: Create in-memory config
+        run: |
+          mkdir -p rust/server
+          cat > rust/server/vss-server-config.toml <<EOF
+          [server_config]
+          host = "127.0.0.1"
+          port = 8080
+          store_type = "in_memory"
+          EOF
+
+      - name: Build server
+        working-directory: rust
+        run: cargo build --release --bin server
+
+      - name: Start VSS server
+        working-directory: rust
+        run: |
+          ./target/release/server vss-server-config.toml > server.log 2>&1 &
+          echo "Server PID: $!"
+
+      - name: Wait for server
+        run: |
+          for i in {1..15}; do
+            if curl -s http://127.0.0.1:8080 > /dev/null; then
+              echo "Server is up!"
+              exit 0
+            fi
+            sleep 1
+          done
+          echo "Server failed:"
+          cat rust/server.log
+          exit 1
+
+      - name: HTTP Smoke Test
+        run: |
+          # PUT: store="test", key="k1", value="kv1"
+          curl -f \
+               -H "Authorization: Bearer test_user" \
+               --data-binary @<(echo "0A04746573741A150A026B3110FFFFFFFFFFFFFFFFFF011A046B317631" | xxd -r -p) \
+               http://127.0.0.1:8080/vss/putObjects
+
+          # GET: store="test", key="k1"
+          RESPONSE=$(curl -f \
+               -H "Authorization: Bearer test_user" \
+               --data-binary @<(echo "0A047465737412026B31" | xxd -r -p) \
+               http://127.0.0.1:8080/vss/getObject)
+
+      - name: Run unit tests
+        working-directory: rust
+        run: cargo test --package impls --lib -- --nocapture

rust/README.md

@@ -25,7 +25,10 @@ cargo build --release
    cargo run -- server/vss-server-config.toml
    ```
 
-   **Note:** For testing purposes you can edit `vss-server-config.toml` to use `store_type` as in-memory instead of PostgreSQL: `store_type = "in_memory"`
+   **Note:** For testing purposes, you can pass `--in-memory` to use in-memory instead of PostgreSQL 
+   ```
+   cargo run -- server/vss-server-config.toml --in-memory
+   ```
 4. VSS endpoint should be reachable at `http://localhost:8080/vss`.
 
 ### Configuration

rust/impls/src/in_memory_store.rs

@@ -288,61 +288,82 @@ impl KvStore for InMemoryBackendImpl {
 		&self, user_token: String, request: ListKeyVersionsRequest,
 	) -> Result<ListKeyVersionsResponse, VssError> {
 		let store_id = request.store_id;
-		let key_prefix = request.key_prefix.unwrap_or("".to_string());
-		let page_token = request.page_token.unwrap_or("".to_string());
+		let key_prefix = request.key_prefix.unwrap_or_default();
+		let page_token = request.page_token.unwrap_or_default();
 		let page_size = request.page_size.unwrap_or(i32::MAX);
 		let limit = std::cmp::min(page_size, LIST_KEY_VERSIONS_MAX_PAGE_SIZE) as usize;
 
+		// Global version only on first page
 		let mut global_version = None;
 		if page_token.is_empty() {
-			let get_global_version_request = GetObjectRequest {
+			let get_global = GetObjectRequest {
 				store_id: store_id.clone(),
 				key: GLOBAL_VERSION_KEY.to_string(),
 			};
-			let get_response = self.get(user_token.clone(), get_global_version_request).await?;
-			global_version = Some(get_response.value.unwrap().version);
+			let resp = self.get(user_token.clone(), get_global).await?;
+			global_version = resp.value.map(|kv| kv.version);
 		}
 
-		let key_versions: Vec<KeyValue> = {
-			let guard = self.store.lock().unwrap();
-			let mut key_versions: Vec<KeyValue> = guard
-				.iter()
-				.filter(|(k, _)| {
-					let parts: Vec<&str> = k.split('#').collect();
-					if parts.len() < 3 {
-						return false;
+		let guard = self.store.lock().unwrap();
+
+		let mut latest_per_key: std::collections::HashMap<String, KeyValue> =
+			std::collections::HashMap::new();
+		for (k, r) in guard.iter() {
+			let parts: Vec<&str> = k.split('#').collect();
+			if parts.len() != 3
+				|| parts[0] != user_token.as_str()
+				|| parts[1] != store_id.as_str()
+				|| parts[2] == GLOBAL_VERSION_KEY
+				|| !parts[2].starts_with(&key_prefix)
+			{
+				continue;
+			}
+			let key = parts[2].to_string();
+			let candidate = KeyValue { key: key.clone(), value: Bytes::new(), version: r.version };
+			latest_per_key
+				.entry(key)
+				.and_modify(|e| {
+					if r.version > e.version {
+						*e = candidate.clone()
 					}
-					parts[0] == user_token.as_str()
-						&& parts[1] == store_id.as_str()
-						&& parts[2].starts_with(&key_prefix)
-						&& parts[2] > page_token.as_str()
-						&& parts[2] != GLOBAL_VERSION_KEY
-				})
-				.map(|(_, record)| KeyValue {
-					key: record.key.clone(),
-					value: Bytes::new(),
-					version: record.version,
 				})
-				.collect();
+				.or_insert(candidate);
+		}
 
-			key_versions.sort_by(|a, b| a.key.cmp(&b.key));
-			key_versions.into_iter().take(limit).collect()
-		};
+		let mut keys: Vec<String> = latest_per_key.keys().cloned().collect();
+		keys.sort();
 
-		let next_page_token = if key_versions.len() == limit {
-			key_versions.last().map(|kv| kv.key.clone())
+		let start_idx = if page_token.is_empty() {
+			0
 		} else {
-			None
+			match keys.iter().position(|k| k == &page_token) {
+				Some(i) => i + 1,
+				None => {
+					return Ok(ListKeyVersionsResponse {
+						key_versions: vec![],
+						next_page_token: None,
+						global_version,
+					});
+				},
+			}
 		};
 
-		Ok(ListKeyVersionsResponse { key_versions, next_page_token, global_version })
+		let page_keys: Vec<String> = keys.into_iter().skip(start_idx).take(limit).collect();
+		let page: Vec<KeyValue> =
+			page_keys.iter().filter_map(|k| latest_per_key.get(k).cloned()).collect();
+
+		let next_page_token =
+			if page.is_empty() { None } else { page.last().map(|kv| kv.key.clone()) };
+
+		Ok(ListKeyVersionsResponse { key_versions: page, next_page_token, global_version })
 	}
 }
 
 #[cfg(test)]
 mod tests {
 	use super::*;
 	use api::define_kv_store_tests;
+	use api::types::{GetObjectRequest, KeyValue, PutObjectRequest};
 	use bytes::Bytes;
 	use tokio::test;
 

rust/server/src/main.rs

@@ -29,19 +29,28 @@ pub(crate) mod vss_service;
 
 fn main() {
 	let args: Vec<String> = std::env::args().collect();
-	if args.len() != 2 {
-		eprintln!("Usage: {} <config-file-path>", args[0]);
+	if args.len() < 2 {
+		eprintln!("Usage: {} <config-file-path> [--in-memory]", args[0]);
 		std::process::exit(1);
 	}
 
-	let config = match util::config::load_config(&args[1]) {
+	let config_path = &args[1];
+	let use_in_memory = args.contains(&"--in-memory".to_string());
+
+	let mut config = match util::config::load_config(config_path) {
 		Ok(cfg) => cfg,
 		Err(e) => {
 			eprintln!("Failed to load configuration: {}", e);
 			std::process::exit(1);
 		},
 	};
 
+	// Override the `store_type` if --in-memory flag passed
+	if use_in_memory {
+		println!("Overriding backend type: using in-memory backend (via --in-memory flag)");
+		config.server_config.store_type = "in_memory".to_string();
+	}
+
 	let addr: SocketAddr =
 		match format!("{}:{}", config.server_config.host, config.server_config.port).parse() {
 			Ok(addr) => addr,