Fix?

test · test · commit 5f828aed13bc · 2025-04-25T15:49:57.000+03:00
diff --git a/Dockerfile b/Dockerfile
@@ -6,25 +6,28 @@
 FROM rust:1.86.0 AS builder
 WORKDIR /app
 
-# Install build dependencies, including OpenSSL development package.
+# Install build deps: pkg-config + OpenSSL headers + a linker
 RUN apt-get update && apt-get install -y \
     pkg-config \
     libssl-dev \
     clang \
     lld \
-    && rm -rf /var/lib/apt/lists/*
+  && rm -rf /var/lib/apt/lists/*
 
-# Copy Cargo manifest and lockfile for dependency caching.
+# Copy only manifests to leverage Docker cache
 COPY Cargo.toml Cargo.lock ./
 
-# Copy the source files.
+# Copy source + any keys
 COPY src ./src
+COPY encryption.key .
 
-# Build the project in release mode.
+# Tell openssl-sys to stop vendoring OpenSSL and to use pkg-config instead
+ENV OPENSSL_NO_VENDOR=1
+
+# Build
 RUN cargo build --release --locked
 
-# Rename the produced binary.
-# (Assuming your Cargo project name is "DynaRust" so that the binary is named "DynaRust" by default)
+# Copy binary for next stage
 RUN cp target/release/DynaRust /server
 
 ###############################
@@ -33,19 +36,13 @@ RUN cp target/release/DynaRust /server
 FROM debian:bookworm-slim
 WORKDIR /app
 
-# Install the OpenSSL runtime library (which provides libssl.so.3).
+# Install just the runtime libssl
 RUN apt-get update && apt-get install -y \
     libssl3 \
-    && rm -rf /var/lib/apt/lists/*
+  && rm -rf /var/lib/apt/lists/*
 
-# Copy the built server binary from the builder stage.
+# Pull in the server
 COPY --from=builder /server /usr/local/bin/server
 
-# Expose port 6660.
 EXPOSE 6660
-
-# Set the entrypoint.
-# By default, the container will run:
-#   /usr/local/bin/server 0.0.0.0:6660
-# Any extra arguments passed to 'docker run' will be appended to the command.
 ENTRYPOINT ["/usr/local/bin/server", "0.0.0.0:6660"]
diff --git a/src/main.rs b/src/main.rs
@@ -20,6 +20,7 @@ use storage::engine::{
 };
 use storage::persistance::{cold_save, load_all_tables};
 use network::broadcaster::{membership_sync, heartbeat, get_membership, update_membership};
+use crate::storage::engine::put_value_internal;
 use crate::storage::snapshoting::start_snapshot_task;
 use crate::storage::statistics::{get_stats, MetricsCollector, MetricsMiddleware};
 use crate::storage::subscription::SubscriptionManager;
@@ -222,6 +223,10 @@ async fn main() -> std::io::Result<()> {
                     .route("/membership", web::get().to(get_membership))
                     .route("/update_membership", web::post().to(update_membership))
                     .route("/heartbeat", web::get().to(heartbeat))
+                    .route(
+                        "/internal/{table}/key/{key}",
+                        web::put().to(put_value_internal),
+                    )
                     // Key–value endpoints with multi‑table support.
                     .route("/{table}/key/{key}", web::get().to(get_value))
                     .route("/{table}/key/{key}", web::put().to(put_value))
@@ -260,6 +265,10 @@ async fn main() -> std::io::Result<()> {
                     .route("/membership", web::get().to(get_membership))
                     .route("/update_membership", web::post().to(update_membership))
                     .route("/heartbeat", web::get().to(heartbeat))
+                    .route(
+                        "/internal/{table}/key/{key}",
+                        web::put().to(put_value_internal),
+                    )
                     // Key–value endpoints with multi‑table support.
                     .route("/{table}/key/{key}", web::get().to(get_value))
                     .route("/{table}/key/{key}", web::put().to(put_value))
diff --git a/src/storage/engine.rs b/src/storage/engine.rs
@@ -1,3 +1,5 @@
+use tokio::sync::Semaphore;
+use crate::Arc;
 use actix_web::{web, HttpRequest, HttpResponse, Responder};
 use futures::future::BoxFuture;
 use serde::{Deserialize, Serialize};
@@ -285,59 +287,35 @@ pub async fn get_value(
     HttpResponse::Ok().json(latest)
 }
 
-/// PUT handler: Inserts or updates a key–value pair with ownership verification.
-/// For external requests, a valid JWT token in the Authorization header is required.
+// external PUT handler
+#[allow(clippy::too_many_arguments)]
 pub async fn put_value(
     req: HttpRequest,
     path: web::Path<(String, String)>,
     state: web::Data<AppState>,
     cluster_data: web::Data<ClusterData>,
     current_addr: web::Data<String>,
     sub_manager: web::Data<SubscriptionManager>,
+    client: web::Data<reqwest::Client>,
+    sem: web::Data<Arc<Semaphore>>,
     body: web::Json<HashMap<String, Value>>,
 ) -> impl Responder {
     let (table_name, key_val) = path.into_inner();
 
-    // Internal replication requests bypass authentication.
-    if req.headers().contains_key("X-Internal-Request") {
-        if env::var("CLUSTER_SECRET").unwrap_or_default().as_str() == req.headers().get("SECRET").unwrap().to_str().unwrap() {
-            let new_value = {
-                let mut store = state.store.write().await;
-                let table = store.entry(table_name.clone()).or_insert_with(HashMap::new);
-                if let Some(existing) = table.get_mut(&key_val) {
-                    existing.update(body.0.clone());
-                    existing.clone()
-                } else {
-                    let user_header = req.headers().get("User").unwrap().to_str().unwrap();
-
-                    // For internal requests, set owner as "internal"
-                    let v = VersionedValue::new(body.0.clone(), String::from(user_header));
-                    table.insert(key_val.clone(), v.clone());
-                    v
-                }
-            };
-            sub_manager
-                .notify(&table_name, &key_val, KeyEvent::Updated(new_value.clone()))
-                .await;
-            return HttpResponse::Created().json(new_value);
-        } else {
-            return HttpResponse::Unauthorized().json(json!({}))
-        }
-    }
-
-    // External requests require a valid JWT token.
+    // 1️⃣ Authenticate external user via JWT
     let user = match extract_user_from_token(&req) {
         Ok(u) => u,
-        Err(err_resp) => return err_resp,
+        Err(resp) => return resp,
     };
 
+    // 2️⃣ Apply local update with ownership check
     let new_value = {
-        let mut store = state.store.write().await;
-        let table = store.entry(table_name.clone()).or_insert_with(HashMap::new);
+        let mut db = state.store.write().await;
+        let table = db.entry(table_name.clone()).or_default();
+
         if let Some(existing) = table.get_mut(&key_val) {
-            // Only allow the owner to update this record.
             if existing.owner != user {
-                return HttpResponse::Unauthorized().json(json!({
+                return HttpResponse::Unauthorized().json(serde_json::json!({
                     "error": "Not authorized to update this record"
                 }));
             }
@@ -350,50 +328,91 @@ pub async fn put_value(
         }
     };
 
+    // 3️⃣ Notify any SSE subscriptions
     sub_manager
         .notify(&table_name, &key_val, KeyEvent::Updated(new_value.clone()))
         .await;
 
-    // Replicate to other nodes.
-    let cluster_guard = cluster_data.nodes.read().await;
-    let active_nodes = get_active_nodes(&*cluster_guard);
-    drop(cluster_guard);
-
-    let replication_factor = active_nodes.len();
-    let targets = get_replication_nodes(&key_val, &active_nodes, replication_factor);
+    // 4️⃣ Replicate *full* VersionedValue to other nodes, but cap concurrency
+    let cluster = cluster_data.nodes.read().await;
+    let active = get_active_nodes(&*cluster);
+    drop(cluster);
 
-    let client = reqwest::Client::new();
-    let mut replication_futures = Vec::new();
+    let targets = get_replication_nodes(&key_val, &active, active.len());
+    let secret = env::var("CLUSTER_SECRET").unwrap_or_default();
 
     for target in targets {
-        if target != *current_addr.get_ref() {
-            let user2 = user.clone();
-            let url = format!("http://{}/{}/key/{}", target, table_name, key_val);
-            let client_clone = client.clone();
-            let value_clone = new_value.clone();
-            let fut = async move {
-                let result = client_clone
-                    .put(&url)
-                    .header("X-Internal-Request", "true")
-                    .header("User", user2.clone())
-                    .header("SECRET", env::var("CLUSTER_SECRET").unwrap_or_default())
-                    .json(&value_clone.value)  // HERE'S THE FIX: Send only the inner value data
-                    .timeout(std::time::Duration::from_secs(3))
-                    .send()
-                    .await;
-                if let Err(e) = result {
-                    println!("Replication error to {}: {}", target, e);
-                }
-            };
-            replication_futures.push(fut);
+        if target == *current_addr.get_ref() {
+            continue;
         }
+        // acquire a permit (will await if > 20 in flight)
+        let permit = <Arc<Semaphore> as Clone>::clone(&sem).acquire_owned().await.unwrap();
+        let cli = client.clone();
+        let payload = new_value.clone();
+        let url = format!(
+            "http://{}/internal/{}/key/{}",
+            target, table_name, key_val
+        );
+
+        let secret_clone = secret.clone();
+
+        tokio::spawn(async move {
+            let _permit = permit; // held until this future ends
+            if let Err(e) = cli
+                .put(&url)
+                .header("X-Internal-Request", "true")
+                .header("SECRET", &secret_clone)
+                .json(&payload)
+                .send()
+                .await
+            {
+                eprintln!("replication to {} failed: {}", target, e);
+            }
+        });
     }
-    tokio::spawn(async move {
-        futures_util::future::join_all(replication_futures).await;
-    });
+
     HttpResponse::Created().json(new_value)
 }
 
+// internal PUT handler (must be mounted at: /internal/{table}/key/{key})
+pub async fn put_value_internal(
+    req: HttpRequest,
+    path: web::Path<(String, String)>,
+    state: web::Data<AppState>,
+    sub_manager: web::Data<SubscriptionManager>,
+    body: web::Json<VersionedValue>,
+) -> impl Responder {
+    let (table_name, key_val) = path.into_inner();
+
+    // 1️⃣ Cluster‐secret check
+    let cluster_secret = env::var("CLUSTER_SECRET").unwrap_or_default();
+    match req.headers().get("SECRET") {
+        Some(h) if h.to_str().unwrap_or("") == cluster_secret => {}
+        _ => return HttpResponse::Unauthorized().finish(),
+    }
+
+    // 2️⃣ Merge the incoming VersionedValue by version number
+    let incoming = body.into_inner();
+    let mut db = state.store.write().await;
+    let table = db.entry(table_name.clone()).or_default();
+
+    let cached = table
+        .entry(key_val.clone())
+        .and_modify(|existing| {
+            if incoming.version > existing.version {
+                *existing = incoming.clone();
+            }
+        })
+        .or_insert_with(|| incoming.clone())
+        .clone();
+
+    // 3️⃣ Notify subscribers
+    sub_manager
+        .notify(&table_name, &key_val, KeyEvent::Updated(cached.clone()))
+        .await;
+
+    HttpResponse::Created().json(cached)
+}
 
 /// DELETE handler: Removes a key with an ownership check.
 pub async fn delete_value(
