Real-time updates

Author: test

Cargo.lock

@@ -14,4 +14,5 @@ base64 = "0.21"
 once_cell = "1.21.3"
 regex = "1.11.1"
 futures = "0.3.31"
+async-stream = "0.3.6"
 

Cargo.toml

@@ -18,6 +18,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::subscription::SubscriptionManager;
 
 /// Declare APP_STATE globally so that it’s available throughout the module.
 static APP_STATE: OnceCell<web::Data<AppState>> = OnceCell::new();
@@ -169,12 +170,14 @@ async fn main() -> std::io::Result<()> {
     let cluster_clone = cluster_data.clone();
     let current_clone = current_node.clone();
     tokio::spawn(membership_sync(cluster_clone, current_clone, 60));
+    let subscription_manager = web::Data::new(SubscriptionManager::new());
 
     println!("Starting distributed DB engine at http://{}", current_node);
 
     // Build and run the HTTP server.
     HttpServer::new(move || {
         App::new()
+            .app_data(subscription_manager.clone())
             .app_data(state.clone())
             .app_data(cluster_data.clone())
             .app_data(web::Data::new(current_node.clone()))
@@ -194,6 +197,9 @@ async fn main() -> std::io::Result<()> {
             // Endpoints to get keys from a table.
             .route("/{table}/keys", web::get().to(get_all_keys))
             .route("/{table}/keys", web::post().to(get_multiple_keys))
+            .route("/{table}/subscribe/{key}", web::get().to(
+                storage::subscription::subscribe_to_key
+            ))
     })
         .bind(bind_addr.as_str())?
         .run()

README.md

@@ -9,7 +9,7 @@ use std::env;
 use std::time::{SystemTime, UNIX_EPOCH};
 use tokio::sync::RwLock;
 use web::Json;
-
+use crate::storage::subscription::{KeyEvent, SubscriptionManager};
 //
 // CORE DATA STRUCTURES
 //
@@ -237,18 +237,21 @@ pub async fn get_value(
     HttpResponse::Ok().json(latest)
 }
 
-/// PUT handler: Inserts or updates a key-value pair and replicates to active nodes
+/// PUT handler: Inserts or updates a key–value pair and replicates to active nodes.
+/// Also sends a live update notification via the SubscriptionManager.
 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>,
+    // New dependency injection for the subscription manager.
+    sub_manager: web::Data<SubscriptionManager>,
     body: web::Json<HashMap<String, Value>>,
 ) -> impl Responder {
     let (table_name, key_val) = path.into_inner();
 
-    // Handle internal replication request (no further replication)
+    // Handle internal replication request (skip notifications).
     if req.headers().contains_key("X-Internal-Request") {
         let new_value = {
             let mut store = state.store.write().await;
@@ -262,10 +265,13 @@ pub async fn put_value(
                 v
             }
         };
+        sub_manager.notify(&table_name, &key_val, KeyEvent::Updated(new_value.clone())).await;
+
         return HttpResponse::Created().json(new_value);
     }
 
-    // For external requests, update locally first
+
+    // External request: update local store.
     let new_value = {
         let mut store = state.store.write().await;
         let table = store.entry(table_name.clone()).or_insert_with(HashMap::new);
@@ -279,7 +285,11 @@ pub async fn put_value(
         }
     };
 
-    // Replicate to other nodes
+    // Notify all subscribers that the key has been updated.
+    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);
@@ -289,7 +299,6 @@ pub async fn put_value(
 
     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);
@@ -314,35 +323,40 @@ pub async fn put_value(
         }
     }
 
-    // Fire replication requests, but don't wait for them to complete
+    // Fire replication requests asynchronously.
     tokio::spawn(async move {
         futures_util::future::join_all(replication_futures).await;
     });
 
-    // Return immediately after local update
     HttpResponse::Created().json(new_value)
 }
 
-/// DELETE handler: Removes a key and replicates the deletion to active nodes
+/// DELETE handler: Removes a key and replicates the deletion to active nodes.
+/// Also notifies subscribers of the deletion event.
 pub async fn delete_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>,
 ) -> impl Responder {
     let (table_name, key_val) = path.into_inner();
 
-    // Handle internal replication request
+    // Internal replication request.
     if req.headers().contains_key("X-Internal-Request") {
         let mut store = state.store.write().await;
         if let Some(table) = store.get_mut(&table_name) {
             table.remove(&key_val);
         }
+        sub_manager
+            .notify(&table_name, &key_val, KeyEvent::Deleted)
+            .await;
+
         return HttpResponse::Ok().json(json!({"message": "Deleted"}));
     }
 
-    // For external requests, delete locally first
+    // External request: remove the key locally.
     let local_status = {
         let mut store = state.store.write().await;
         if let Some(table) = store.get_mut(&table_name) {
@@ -357,7 +371,12 @@ pub async fn delete_value(
             .to_string()
     };
 
-    // Replicate deletion to other nodes
+    // Notify subscribers that the key has been deleted.
+    sub_manager
+        .notify(&table_name, &key_val, KeyEvent::Deleted)
+        .await;
+
+    // Replicate deletion to other nodes.
     let cluster_guard = cluster_data.nodes.read().await;
     let active_nodes = get_active_nodes(&*cluster_guard);
     drop(cluster_guard);
@@ -367,12 +386,10 @@ pub async fn delete_value(
 
     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 {
                 let result = client_clone
                     .delete(&url)
@@ -385,20 +402,16 @@ pub async fn delete_value(
                     println!("Deletion replication error to {}: {}", target, e);
                 }
             };
-
             replication_futures.push(fut);
         }
     }
 
-    // Fire replication requests in background
     tokio::spawn(async move {
         futures_util::future::join_all(replication_futures).await;
     });
 
-    // Return immediately after local deletion
     HttpResponse::Ok().json(json!({"message": local_status}))
 }
-
 //
 // TABLE MANAGEMENT HANDLERS
 //

src/main.rs

@@ -1,2 +1,3 @@
 pub mod engine;
-pub mod persistance;
+pub mod persistance;
+pub(crate) mod subscription;

src/storage/engine.rs

@@ -0,0 +1,93 @@
+// storage/subscription.rs
+
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+use std::sync::Arc;
+use tokio::sync::{broadcast, RwLock};
+
+/// An event describing a key update (or deletion).
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub enum KeyEvent {
+    Updated(VersionedValue),
+    Deleted,
+}
+
+/// Manages broadcast channels for keys so that multiple subscribers can listen.
+#[derive(Clone)]
+pub struct SubscriptionManager {
+    // The key is a combination of table name and key (e.g. "default/mykey")
+    pub channels: Arc<RwLock<HashMap<String, broadcast::Sender<KeyEvent>>>>,
+}
+
+impl SubscriptionManager {
+    pub fn new() -> Self {
+        Self {
+            channels: Arc::new(RwLock::new(HashMap::new())),
+        }
+    }
+
+    /// Subscribe to updates for a given table/key.
+    pub async fn subscribe(&self, table: &str, key: &str) -> broadcast::Receiver<KeyEvent> {
+        let identifier = format!("{}/{}", table, key);
+        let mut channels = self.channels.write().await;
+        if let Some(sender) = channels.get(&identifier) {
+            sender.subscribe()
+        } else {
+            // Create a new channel if none exists; here we use a buffer of 16 events.
+            let (tx, rx) = broadcast::channel(16);
+            channels.insert(identifier, tx);
+            rx
+        }
+    }
+
+    /// Send a notification for the given table/key.
+    pub async fn notify(&self, table: &str, key: &str, event: KeyEvent) {
+        let identifier = format!("{}/{}", table, key);
+        let channels = self.channels.read().await;
+        if let Some(sender) = channels.get(&identifier) {
+            // Ignore send error if there are no subscribers
+            let _ = sender.send(event);
+        }
+    }
+}
+// storage/subscription.rs (continued)
+
+use actix_web::{web, HttpResponse, Responder};
+use futures::stream::StreamExt;
+use crate::storage::engine::VersionedValue;
+
+/// Subscribe to updates on a key. The URL contains the table and key.
+pub async fn subscribe_to_key(
+    path: web::Path<(String, String)>,
+    sub_manager: web::Data<SubscriptionManager>,
+) -> impl Responder {
+    let (table, key) = path.into_inner();
+
+    // Get a broadcast receiver for this key.
+    let mut rx = sub_manager.subscribe(&table, &key).await;
+
+    // Build an SSE stream. Each event is yielded as a chunk in the HTTP response.
+    let stream = async_stream::stream! {
+        loop {
+            match rx.recv().await {
+                Ok(event) => {
+                    // Serialize the event as JSON.
+                    let data = serde_json::to_string(&event)
+                        .unwrap_or_else(|_| "null".to_string());
+                    // SSE events MUST be formatted as "data: <payload>\n\n".
+                    yield Ok::<_, actix_web::Error>(
+                        web::Bytes::from(format!("data: {}\n\n", data))
+                    );
+                }
+                // If messages are dropped, simply continue.
+                Err(broadcast::error::RecvError::Lagged(_)) => continue,
+                // Exit the loop when the sender is dropped.
+                Err(broadcast::error::RecvError::Closed) => break,
+            }
+        }
+    };
+
+    HttpResponse::Ok()
+        .append_header(("Content-Type", "text/event-stream"))
+        .streaming(stream)
+}