Patch value

Author: Istrate Andrei-Eduard

.DS_Store

@@ -0,0 +1,30 @@
+# DynaRust Project
+
+## Project Overview
+DynaRust is a distributed key-JSON-value datastore built in Rust. It provides in-memory caching, on-disk persistence, background replication for eventual consistency, and real-time update capabilities using Server-Sent Events (SSE). It handles cluster membership dynamically, allowing nodes to join and synchronize their state.
+
+## Building and Running
+*   **Build the project:**
+    ```bash
+    cargo build --release
+    ```
+*   **Run a node:**
+    ```bash
+    # Run the first node
+    ./target/release/DynaRust <LISTEN_ADDRESS>
+
+    # Run additional nodes and join the cluster
+    ./target/release/DynaRust <LISTEN_ADDRESS> <JOIN_ADDRESS>
+    ```
+
+## Development Conventions
+*   **Rust & Actix-Web:** Built predominantly with Rust and Actix-Web for high-performance concurrent request handling.
+*   **Authentication:** Endpoints (except GET) require JWT token via `Authorization: Bearer <token>`. Node-to-node replication uses a cluster secret defined in `.env`.
+*   **Concurrency:** Utilizes `tokio` for async tasks, e.g., snapshots, gossip membership sync, and cluster replication.
+*   **Storage Architecture:** Data is stored in-memory using `HashMap<String, HashMap<String, VersionedValue>>` with background disk persistence.
+
+## Key Files
+*   `src/main.rs`: Entry point and Actix-Web server initialization.
+*   `src/storage/engine.rs`: Core data structures (`VersionedValue`, `AppState`) and CRUD handler logic (`put_value`, `get_value`, `delete_value`).
+*   `src/network/broadcaster.rs`: Cluster membership gossip protocol implementation.
+*   `src/storage/subscription.rs`: Real-time updates handler using Server-Sent Events (SSE).

GEMINI.md

@@ -110,22 +110,22 @@ All operations except **GET** require a valid JWT in the `Authorization: Bearer
 
 2. **✍️ Store or Update a Value (HTTP PUT)**  
    Create or update a value under `{table}/{key}`.  
-   - URL: `/default/key/mykey`  
+   - URL: `/{table}/key/{key}`  
    - Headers:  
      ```
      Content-Type: application/json  
      Authorization: Bearer <JWT‑TOKEN>
      ```  
    - Body:  
      ```json
-     { "value": "mydata" }
+     { "value": { ... } }
      ```  
    - Success:  
      • `201 Created`  
      • Body (VersionedValue):  
        ```json
        {
-         "value": "mydata",
+         "value": { ... },
          "version": 1,
          "timestamp": 1618880821123,
          "owner": "alice"
@@ -134,25 +134,52 @@ All operations except **GET** require a valid JWT in the `Authorization: Bearer
    - Errors:  
      • `401 Unauthorized` if missing/invalid JWT or not owner on update  
 
-3. **🔍 Retrieve a Value (HTTP GET)**  
+3. **🛠️ Partially Update a Value (HTTP PATCH)**  
+   Merge updates into an existing value under `{table}/{key}`. Only the owner can patch.
+   - URL: `/{table}/key/{key}`  
+   - Headers:  
+     ```
+     Content-Type: application/json  
+     Authorization: Bearer <JWT‑TOKEN>
+     ```  
+   - Body:  
+     ```json
+     { "new_field": "updated_data" }
+     ```  
+   - Success:  
+     • `200 OK`  
+     • Body (Updated VersionedValue):  
+       ```json
+       {
+         "value": { "original_field": "...", "new_field": "updated_data" },
+         "version": 2,
+         "timestamp": 1618880825000,
+         "owner": "alice"
+       }
+       ```  
+   - Errors:  
+     • `401 Unauthorized` if missing/invalid JWT or not owner  
+     • `404 Not Found` if key/table missing  
+
+4. **🔍 Retrieve a Value (HTTP GET)**  
    Anyone can fetch a key’s latest value.  
-   - URL: `/default/key/mykey`  
+   - URL: `/{table}/key/{key}`  
    - Success:  
      • `200 OK`  
      • Body:  
        ```json
        {
-         "value": "mydata",
+         "value": { ... },
          "version": 1,
          "timestamp": 1618880821123,
          "owner": "alice"
        }
        ```  
      • `404 Not Found` if key/table missing  
 
-4. **🗑️ Delete a Value (HTTP DELETE)**  
+5. **🗑️ Delete a Value (HTTP DELETE)**  
    Only the owner may delete.  
-   - URL: `/default/key/mykey`  
+   - URL: `/{table}/key/{key}`  
    - Header:  
      ```
      Authorization: Bearer <JWT‑TOKEN>
@@ -167,46 +194,46 @@ All operations except **GET** require a valid JWT in the `Authorization: Bearer
      • `401 Unauthorized` if no JWT or not owner  
      • `404 Not Found` if key/table missing  
 
-5. **📚 Fetch Entire Table Store (HTTP GET)**  
+6. **📚 Fetch Entire Table Store (HTTP GET)**  
    List all key→VersionedValue pairs in a table.  
-   - URL: `/default/store`  
+   - URL: `/{table}/store`  
    - Success:  
      • `200 OK`  
      • Body:  
        ```json
        {
-         "key1": { "value":"v1","version":2,…,"owner":"bob" },
+         "key1": { "value":{...},"version":2,…,"owner":"bob" },
          "key2": { … }
        }
        ```  
    - `404 Not Found` if table missing  
 
-6. **🔑 List or Batch‑Fetch Keys**  
-   6.1 **GET** `/default/keys`  
+7. **🔑 List or Batch‑Fetch Keys**  
+   7.1 **GET** `/{table}/keys`  
        • `200 OK` →  
          ```json
          ["key1","key2",…]
          ```  
-   6.2 **POST** `/default/keys`  
+   7.2 **POST** `/{table}/keys`  
        - Body:  
          ```json
          ["key1","key2","key3"]
          ```  
        - `200 OK` →  
          ```json
          {
-           "key1": { "value":"v1","version":… },
+           "key1": { "value":{...},"version":… },
            "key3": { … }
          }
          ```  
        (non‑existent keys are omitted)
 
-7. **🔔 Subscribe to Real‑Time Updates (SSE)**  
+8. **🔔 Subscribe to Real‑Time Updates (SSE)**  
    Instant updates on a single key.  
-   - URL: `/default/subscribe/mykey`  
+   - URL: `/{table}/subscribe/{key}`  
    - Usage:  
      ```bash
-     curl -N http://localhost:6660/default/subscribe/mykey
+     curl -N http://localhost:6660/{table}/subscribe/{key}
      ```  
    - Each update is a JSON event:  
      ```json
@@ -232,22 +259,26 @@ TOKEN=$(curl -s -X POST http://localhost:6660/auth/alice \
 curl -i -X PUT http://localhost:6660/default/key/foo \
   -H "Content-Type: application/json" \
   -H "Authorization: Bearer $TOKEN" \
-  -d '{"value":"hello"}'
+  -d '{"value": {"name": "bar"}}'
+
+# 4) PATCH (partially update)
+curl -i -X PATCH http://localhost:6660/default/key/foo \
+  -H "Content-Type: application/json" \
+  -H "Authorization: Bearer $TOKEN" \
+  -d '{"age": 30}'
 
-# 4) GET
+# 5) GET
 curl -i http://localhost:6660/default/key/foo
 
-# 5) DELETE (owner only)
+# 6) DELETE (owner only)
 curl -i -X DELETE http://localhost:6660/default/key/foo \
   -H "Authorization: Bearer $TOKEN"
 
-# 6) Node stats
+# 7) Node stats
 curl http://localhost:6660/stats
-Response:
-{"tables":{"auth":2,"test":20088,"default":1},"total_keys":20091,"total_requests":5,"average_latency_ms":0.4171416,"active_sse_connections":0}
 ```
 
-> ⚠️ PUT/DELETE without a valid JWT → **401 Unauthorized**  
+> ⚠️ PUT/PATCH/DELETE without a valid JWT → **401 Unauthorized**  
 > 🔍 GET is always open (no auth needed).
 ---
 

README.md

@@ -15,7 +15,7 @@ use std::sync::Arc;
 use tokio::sync::{RwLock, Semaphore};
 use storage::engine::{
     AppState, ClusterData, delete_value, get_value, get_table_store, join_cluster, put_value,
-    get_all_keys, get_multiple_keys, NodeInfo, NodeStatus, current_timestamp,
+    patch_value, get_all_keys, get_multiple_keys, NodeInfo, NodeStatus, current_timestamp,
     VersionedValue, get_global_store
 };
 use storage::persistance::{cold_save, load_all_tables};
@@ -239,6 +239,7 @@ async fn main() -> std::io::Result<()> {
                     // 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))
+                    .route("/{table}/key/{key}", web::patch().to(patch_value))
                     .route("/{table}/key/{key}", web::delete().to(delete_value))
                     // Endpoint to fetch a table’s entire in‑memory store.
                     .route("/{table}/store", web::get().to(get_table_store))
@@ -283,6 +284,7 @@ async fn main() -> std::io::Result<()> {
                     // 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))
+                    .route("/{table}/key/{key}", web::patch().to(patch_value))
                     .route("/{table}/key/{key}", web::delete().to(delete_value))
                     // Endpoint to fetch a table’s entire in‑memory store.
                     .route("/{table}/store", web::get().to(get_table_store))

src/main.rs

@@ -53,6 +53,14 @@ impl VersionedValue {
         self.version += 1;
         self.timestamp = current_timestamp();
     }
+
+    pub fn patch(&mut self, updates: HashMap<String, Value>) {
+        for (k, v) in updates {
+            self.value.insert(k, v);
+        }
+        self.version += 1;
+        self.timestamp = current_timestamp();
+    }
 }
 
 /// The local in-memory database state.
@@ -305,6 +313,93 @@ pub async fn get_value(
 
     HttpResponse::Ok().json(latest)
 }
+// external PATCH handler
+#[allow(clippy::too_many_arguments)]
+pub async fn patch_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();
+
+    // 1️⃣ Authenticate external user via JWT
+    let user = match extract_user_from_token(&req) {
+        Ok(u) => u,
+        Err(resp) => return resp,
+    };
+
+    // 2️⃣ Apply local update with ownership check
+    let new_value = {
+        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) {
+            if existing.owner != user {
+                return HttpResponse::Unauthorized().json(serde_json::json!({
+                    "error": "Not authorized to update this record"
+                }));
+            }
+            existing.patch(body.0.clone());
+            existing.clone()
+        } else {
+            return HttpResponse::NotFound().json(serde_json::json!({
+                "error": "Record not found"
+            }));
+        }
+    };
+
+    // 3️⃣ Notify any SSE subscriptions
+    sub_manager
+        .notify(&table_name, &key_val, KeyEvent::Updated(new_value.clone()))
+        .await;
+
+    // 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 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() {
+            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);
+            }
+        });
+    }
+
+    HttpResponse::Ok().json(new_value)
+}
+
 // external PUT handler
 #[allow(clippy::too_many_arguments)]
 pub async fn put_value(

src/storage/engine.rs

@@ -2,8 +2,8 @@
 
 # Configuration
 PORT=6660
-BASE_URL="http://localhost:$PORT"
-TOTAL_RECORDS=10000  # Change to 100000 for the full test
+BASE_URL="http://linking.studio:$PORT"
+TOTAL_RECORDS=1000  # Change to 100000 for the full test
 CONCURRENCY=50       # Number of parallel requests
 TABLE="test_perf"