FIX

Author: Istrate Andrei-Eduard

src/security/authentication.rs

@@ -192,7 +192,7 @@ pub async fn access(
     }
 
     // Generate JWT token for the authenticated user
-    let exp = (Utc::now() + Duration::hours(1)).timestamp() as usize;
+    let exp = (Utc::now() + Duration::hours(24)).timestamp() as usize;
     let claims = Claims { sub: username.clone(), exp };
 
     let token = match encode(

src/storage/engine.rs

@@ -440,6 +440,9 @@ pub async fn delete_value(
     cluster_data: web::Data<ClusterData>,
     current_addr: web::Data<String>,
     sub_manager: web::Data<SubscriptionManager>,
+    // ADDED: Shared client and semaphore
+    client: web::Data<reqwest::Client>,
+    sem: web::Data<Arc<Semaphore>>,
 ) -> impl Responder {
     let (table_name, key_val) = path.into_inner();
 
@@ -492,7 +495,7 @@ pub async fn delete_value(
         } else {
             "Table not found locally"
         }
-            .to_string()
+        .to_string()
     };
 
     sub_manager
@@ -506,13 +509,16 @@ pub async fn delete_value(
     let replication_factor = active_nodes.len();
     let targets = get_replication_nodes(&key_val, &active_nodes, replication_factor);
 
-    let client = reqwest::Client::new();
-    let mut replication_futures = Vec::new();
     for target in targets {
         if target != *current_addr.get_ref() {
             let url = format!("http://{}/{}/key/{}", target, table_name, key_val);
             let client_clone = client.clone();
-            let fut = async move {
+            
+            // Acquire permit to cap concurrent outgoing replication tasks
+            let permit = <Arc<Semaphore> as Clone>::clone(&sem).acquire_owned().await.unwrap();
+
+            tokio::spawn(async move {
+                let _permit = permit; // Permit is held until the deletion request completes
                 let result = client_clone
                     .delete(&url)
                     .header("X-Internal-Request", "true")
@@ -522,16 +528,12 @@ pub async fn delete_value(
                 if let Err(e) = result {
                     println!("Deletion replication error to {}: {}", target, e);
                 }
-            };
-            replication_futures.push(fut);
+            });
         }
     }
-    tokio::spawn(async move {
-        futures_util::future::join_all(replication_futures).await;
-    });
+    
     HttpResponse::Ok().json(json!({"message": local_status}))
 }
-
 //
 // TABLE MANAGEMENT HANDLERS
 //
@@ -592,6 +594,9 @@ pub async fn get_multiple_keys(
 pub async fn join_cluster(
     cluster: web::Data<ClusterData>,
     request: Json<JoinRequest>,
+    // ADDED: Shared client and semaphore
+    client: web::Data<reqwest::Client>,
+    sem: web::Data<Arc<Semaphore>>,
 ) -> impl Responder {
     let cluster_secret = env::var("CLUSTER_SECRET")
         .unwrap_or_else(|_| "default_secret".to_string());
@@ -603,6 +608,7 @@ pub async fn join_cluster(
     let new_node = request.node.clone();
     let cluster2 = cluster.clone();
     let mut nodes_guard = cluster.nodes.write().await;
+    
     if !nodes_guard.contains_key(&new_node) {
         nodes_guard.insert(
             new_node.clone(),
@@ -612,17 +618,26 @@ pub async fn join_cluster(
             },
         );
 
-        tokio::spawn(async move {
+        // Safely bound the background sleep/gossip task. 
+        // If the system is maxed out, we skip the background gossip rather than queuing unbound tasks.
+        if let Ok(permit) = <Arc<Semaphore> as Clone>::clone(&sem).try_acquire_owned() {
+            tokio::spawn(async move {
+                let _permit = permit; // Permit held during the sleep and gossip phase
                 time::sleep(std::time::Duration::from_secs(5)).await;
+                
                 let args: Vec<String> = env::args().collect();
-                let current_node = args[1].clone();
-                let client = reqwest::Client::new();
-                gossip_membership(&cluster2, &client, &*current_node).await;
-        });
+                if args.len() > 1 {
+                    let current_node = args[1].clone();
+                    gossip_membership(&cluster2, &client, &current_node).await;
+                }
+            });
+        } else {
+            eprintln!("Warning: Dropping gossip broadcast for new node {} due to high load", new_node);
+        }
     }
+    
     HttpResponse::Ok().json(nodes_guard.clone())
 }
-
 pub async fn get_global_store(
     req: HttpRequest,
     state: web::Data<AppState>,

stress.sh

@@ -0,0 +1,80 @@
+#!/bin/bash
+
+# Configuration
+PORT=6660
+BASE_URL="http://localhost:$PORT"
+TOTAL_RECORDS=10000  # Change to 100000 for the full test
+CONCURRENCY=50       # Number of parallel requests
+TABLE="test_perf"
+
+echo "=== Starting Performance Test ($TOTAL_RECORDS records) ==="
+
+# 1) Register & Authenticate Alice
+echo "[1/4] Registering and authenticating user..."
+curl -s -X POST "$BASE_URL/auth/alice" \
+  -H "Content-Type: application/json" \
+  -d '{"secret":"s3cr3t"}' > /dev/null
+
+TOKEN=$(curl -s -X POST "$BASE_URL/auth/alice" \
+  -H "Content-Type: application/json" \
+  -d '{"secret":"s3cr3t"}' | jq -r .token)
+
+if [ -z "$TOKEN" ] || [ "$TOKEN" == "null" ]; then
+    echo "Error: Failed to retrieve JWT token. Is the server running?"
+    exit 1
+fi
+
+# 50-word Lorem Ipsum JSON payload
+PAYLOAD='{"value": "Lorem ipsum dolor sit amet, consectetur adipiscing elit. Sed do eiusmod tempor incididunt ut labore et dolore magna aliqua. Ut enim ad minim veniam, quis nostrud exercitation ullamco laboris nisi ut aliquip ex ea commodo consequat. Duis aute irure dolor in reprehenderit in voluptate velit esse cillum dolore eu fugiat nulla pariatur. Excepteur sint occaecat cupidatat non proident."}'
+
+# 2) Parallel PUT Operations (Data Storage Test)
+echo "[2/4] Inserting $TOTAL_RECORDS records with $CONCURRENCY concurrent workers..."
+START_TIME=$(date +%s)
+
+# We use seq and xargs to blast the server with concurrent requests
+seq 1 $TOTAL_RECORDS | xargs -n 1 -P $CONCURRENCY -I {} \
+  curl -s -X PUT "$BASE_URL/$TABLE/key/req_{}" \
+  -H "Content-Type: application/json" \
+  -H "Authorization: Bearer $TOKEN" \
+  -d "$PAYLOAD" -o /dev/null
+
+END_TIME=$(date +%s)
+echo "Done! Insertion took $((END_TIME - START_TIME)) seconds."
+
+# 3) GET Operation Latency Test
+echo "[3/4] Measuring GET Latency..."
+TOTAL_TIME=0
+SAMPLE_SIZE=100
+
+for i in $(seq 1 $SAMPLE_SIZE); do
+    # Fetch a random key from the ones we just created
+    RANDOM_KEY="req_$((1 + $RANDOM % $TOTAL_RECORDS))"
+    
+    # Use curl's write-out feature to get the exact time in seconds, then convert to ms
+    TIME=$(curl -s -o /dev/null -w "%{time_total}" "$BASE_URL/$TABLE/key/$RANDOM_KEY")
+    TOTAL_TIME=$(awk "BEGIN {print $TOTAL_TIME + $TIME}")
+done
+
+AVG_LATENCY=$(awk "BEGIN {print ($TOTAL_TIME / $SAMPLE_SIZE) * 1000}")
+echo "Average GET Latency over $SAMPLE_SIZE requests: ${AVG_LATENCY} ms"
+
+# 4) System Stats & Memory
+echo "[4/4] Fetching Node Stats..."
+curl -s "$BASE_URL/stats" | jq .
+
+echo "=== System Resource Check ==="
+# Find the PID of the server running on the port
+PID=$(lsof -t -i:$PORT)
+if [ -n "$PID" ]; then
+    echo "Memory Consumption of Server (PID $PID):"
+    ps -p $PID -o %cpu,%mem,rss | awk 'NR==1{print $0; next} {print $1"%  ", $2"%  ", $3/1024 " MB"}'
+else
+    echo "Could not find PID for port $PORT to check memory."
+fi
+
+# Check disk space used by the cold storage (assuming it's in the current directory or a specific data dir)
+# Adjust the path './' if your AppState base_dir is located elsewhere
+echo "Disk Space Used by storage directories:"
+du -sh ./*/ 2>/dev/null | grep -E "auth|test|default|$TABLE" || echo "No data directories found in current path."
+
+echo "=== Test Complete ==="