feat: exclude pending state in broadcast update (#546)

* feat: don't include pending state in broadcast update

* feat: enable large refs for keck server

* feat: put apply update into a separate thread

apps/keck/Cargo.toml

@@ -58,7 +58,7 @@ serde = { workspace = true, features = ["derive"] }
 serde_json = { workspace = true }
 thiserror = { workspace = true }
 
-jwst-core = { workspace = true }
+jwst-core = { workspace = true, features = ["large_refs"] }
 jwst-logger = { workspace = true }
 jwst-rpc = { workspace = true }
 jwst-storage = { workspace = true }

libs/jwst-codec/src/doc/document.rs

@@ -289,7 +289,7 @@ impl Doc {
             }
         }
 
-        store.diff_state_vector(&before_state)
+        store.diff_state_vector(&before_state, false)
     }
 
     pub fn keys(&self) -> Vec<String> {
@@ -356,7 +356,7 @@ impl Doc {
     }
 
     pub fn encode_state_as_update(&self, sv: &StateVector) -> JwstCodecResult<Update> {
-        self.store.read().unwrap().diff_state_vector(sv)
+        self.store.read().unwrap().diff_state_vector(sv, true)
     }
 
     pub fn get_state_vector(&self) -> StateVector {

libs/jwst-codec/src/doc/publisher.rs

@@ -76,7 +76,7 @@ impl DocPublisher {
                         );
 
                         history.resolve_with_store(&store);
-                        let (binary, history) = match store.diff_state_vector(&last_update) {
+                        let (binary, history) = match store.diff_state_vector(&last_update, false) {
                             Ok(update) => {
                                 drop(store);
 

libs/jwst-codec/src/doc/store.rs

@@ -728,7 +728,7 @@ impl DocStore {
         diff
     }
 
-    pub fn diff_state_vector(&self, sv: &StateVector) -> JwstCodecResult<Update> {
+    pub fn diff_state_vector(&self, sv: &StateVector, with_pending: bool) -> JwstCodecResult<Update> {
         let update_structs = Self::diff_structs(&self.items, sv)?;
 
         let mut update = Update {
@@ -737,8 +737,10 @@ impl DocStore {
             ..Update::default()
         };
 
-        if let Some(pending) = &self.pending {
-            Update::merge_into(&mut update, [pending.clone()])
+        if with_pending {
+            if let Some(pending) = &self.pending {
+                Update::merge_into(&mut update, [pending.clone()])
+            }
         }
 
         Ok(update)

libs/jwst-core/src/workspaces/sync.rs

@@ -102,7 +102,9 @@ impl Workspace {
                                     .ok()
                             }),
                             DocMessage::Step2(update) => {
+                                let len = update.len();
                                 if let Ok(update) = Update::read(&mut RawDecoder::new(update)) {
+                                    debug!("step2 apply update: {len}");
                                     if let Err(e) = doc.apply_update(update) {
                                         warn!("failed to apply update: {:?}", e);
                                     }
@@ -118,7 +120,9 @@ impl Workspace {
 
                                     let mut encoder = RawEncoder::default();
                                     update.write(&mut encoder)?;
-                                    Ok(Some(encoder.into_inner()))
+                                    let update = encoder.into_inner();
+                                    debug!("step3 return changed update: {}", update.len());
+                                    Ok(Some(update))
                                 })
                                 .map_err(|e| warn!("failed to apply update: {:?}", e))
                                 .ok()

libs/jwst-rpc/src/broadcast.rs

@@ -45,8 +45,13 @@ pub async fn subscribe(workspace: &Workspace, identifier: String, sender: Broadc
     {
         let sender = sender.clone();
         let workspace_id = workspace.id();
-        workspace.subscribe_doc(move |update, _history| {
-            debug!("workspace {} changed: {}bytes", workspace_id, update.len());
+        workspace.subscribe_doc(move |update, history| {
+            debug!(
+                "workspace {} changed: {}bytes, {} histories",
+                workspace_id,
+                update.len(),
+                history.len()
+            );
 
             match encode_update_with_guid(update.to_vec(), workspace_id.clone())
                 .and_then(|update_with_guid| encode_update_as_message(update.to_vec()).map(|u| (update_with_guid, u)))

libs/jwst-rpc/src/context.rs

@@ -5,10 +5,13 @@ use chrono::Utc;
 use jwst_codec::{CrdtReader, RawDecoder};
 use jwst_core::{DocStorage, Workspace};
 use jwst_storage::{JwstStorage, JwstStorageResult};
-use tokio::sync::{
-    broadcast::{channel as broadcast, error::RecvError, Receiver as BroadcastReceiver, Sender as BroadcastSender},
-    mpsc::{Receiver as MpscReceiver, Sender as MpscSender},
-    Mutex,
+use tokio::{
+    sync::{
+        broadcast::{channel as broadcast, error::RecvError, Receiver as BroadcastReceiver, Sender as BroadcastSender},
+        mpsc::{Receiver as MpscReceiver, Sender as MpscSender},
+        Mutex,
+    },
+    task::spawn_blocking,
 };
 
 use super::{
@@ -155,7 +158,7 @@ pub trait RpcContextImpl<'a> {
         // collect messages from remote
         let identifier = identifier.to_owned();
         let id = id.to_string();
-        let mut workspace = self
+        let workspace = self
             .get_storage()
             .get_workspace(&id)
             .await
@@ -185,7 +188,12 @@ pub trait RpcContextImpl<'a> {
                             let updates = std::mem::take(&mut updates);
                             let updates_len = updates.len();
                             let ts = Instant::now();
-                            let message = workspace.sync_messages(updates);
+                            let message = {
+                                let mut workspace = workspace.clone();
+                                spawn_blocking(move || workspace.sync_messages(updates))
+                                    .await
+                                    .unwrap()
+                            };
                             if ts.elapsed().as_micros() > 50 {
                                 debug!(
                                     "apply {updates_len} remote update cost: {}ms",