refactor(rtc): merge several ws-requests into one init-request

Author: RCmerci

src/main/frontend/worker/rtc/client.cljs

@@ -19,22 +19,62 @@
             [missionary.core :as m]
             [tick.core :as tick]))
 
-(defn- new-task--register-graph-updates
-  [get-ws-create-task graph-uuid major-schema-version repo]
+(defn- apply-remote-updates-from-apply-ops
+  [apply-ops-resp graph-uuid repo conn date-formatter add-log-fn]
+  (if-let [remote-ex (:ex-data apply-ops-resp)]
+    (do (add-log-fn :rtc.log/pull-remote-data (assoc remote-ex :sub-type :pull-remote-data-exception))
+        (case (:type remote-ex)
+          :graph-lock-failed nil
+          :graph-lock-missing
+          (throw r.ex/ex-remote-graph-lock-missing)
+          :rtc.exception/get-s3-object-failed
+          (throw (ex-info (:ex-message apply-ops-resp) (:ex-data apply-ops-resp)))
+          ;;else
+          (throw (ex-info "Unavailable3" {:remote-ex remote-ex}))))
+    (do (assert (pos? (:t apply-ops-resp)) apply-ops-resp)
+        (r.remote-update/apply-remote-update
+         graph-uuid repo conn date-formatter {:type :remote-update :value apply-ops-resp} add-log-fn))))
+
+(defn- new-task--init-request
+  [get-ws-create-task graph-uuid major-schema-version repo conn *last-calibrate-t *server-schema-version add-log-fn]
   (m/sp
-    (try
-      (let [{remote-t :t :as resp}
-            (m/? (ws-util/send&recv get-ws-create-task {:action "register-graph-updates"
-                                                        :graph-uuid graph-uuid
-                                                        :schema-version (str major-schema-version)}))]
-        (rtc-log-and-state/update-remote-t graph-uuid remote-t)
-        (when-not (client-op/get-local-tx repo)
-          (client-op/update-local-tx repo remote-t))
-        resp)
-      (catch :default e
-        (if (= :rtc.exception/remote-graph-not-ready (:type (ex-data e)))
-          (throw (ex-info "remote graph is still creating" {:missionary/retry true} e))
-          (throw e))))))
+    (let [t-before (client-op/get-local-tx repo)
+          get-graph-skeleton? (or (nil? @*last-calibrate-t)
+                                  (< 500 (- t-before @*last-calibrate-t)))]
+      (try
+        (let [t-before (client-op/get-local-tx repo)
+              {remote-t :t
+               server-schema-version :server-schema-version
+               server-builtin-db-idents :server-builtin-db-idents
+               :as resp}
+              (m/? (ws-util/send&recv get-ws-create-task {:action "init-request"
+                                                          :graph-uuid graph-uuid
+                                                          :schema-version (str major-schema-version)
+                                                          :t-before (or t-before 1)
+                                                          :get-graph-skeleton get-graph-skeleton?}))]
+          (if-let [remote-ex (:ex-data resp)]
+            (do
+              (add-log-fn :rtc.log/init-request remote-ex)
+              (case (:type remote-ex)
+                :graph-lock-failed nil
+                :graph-lock-missing (throw r.ex/ex-remote-graph-lock-missing)
+                ;; else
+                (throw (ex-info "Unavailable4" {:remote-ex remote-ex}))))
+            (do
+              (when server-schema-version
+                (reset! *server-schema-version server-schema-version)
+                (reset! *last-calibrate-t remote-t))
+              (when remote-t
+                (rtc-log-and-state/update-remote-t graph-uuid remote-t)
+                (when (not t-before)
+                  (client-op/update-local-tx repo remote-t)))
+              (when (and server-schema-version server-builtin-db-idents)
+                (r.skeleton/calibrate-graph-skeleton server-schema-version server-builtin-db-idents @conn))
+              resp)))
+        (catch :default e
+          (if (= :rtc.exception/remote-graph-not-ready (:type (ex-data e)))
+            (throw (ex-info "remote graph is still creating" {:missionary/retry true} e))
+            (throw e)))))))
 
 (def ^:private *register-graph-updates-sent
   "ws -> [bool, added-inst, [graph-uuid,major-schema-version,repo]]"
@@ -54,8 +94,8 @@
 (defn ensure-register-graph-updates--memoized
   "Return a task: get or create a mws(missionary wrapped websocket).
   see also `ws/get-mws-create`.
-  But ensure `register-graph-updates` and `calibrate-graph-skeleton` has been sent"
-  [get-ws-create-task graph-uuid major-schema-version repo conn
+  But ensure `init-request` and `calibrate-graph-skeleton` has been sent"
+  [get-ws-create-task graph-uuid major-schema-version repo conn date-formatter
    *last-calibrate-t *online-users *server-schema-version add-log-fn]
   (m/sp
     (let [ws (m/? get-ws-create-task)
@@ -81,14 +121,17 @@
                                          10000)))]
                 (reset! *online-users online-users)))
             :succ (constantly nil)))
-        (let [{:keys [max-remote-schema-version]}
+        (let [{:keys [max-remote-schema-version] :as init-request-resp}
               (try
                 (m/?
                  (c.m/backoff
                   {:delay-seq ;retry 5 times if remote-graph is creating (4000 8000 16000 32000 64000)
                    (take 5 (drop 2 c.m/delays))
                    :reset-flow worker-flows/online-event-flow}
-                  (new-task--register-graph-updates get-ws-create-task graph-uuid major-schema-version repo)))
+                  (new-task--init-request
+                   get-ws-create-task graph-uuid major-schema-version repo conn
+                   *last-calibrate-t *server-schema-version
+                   add-log-fn)))
                 (catch :default e
                   (swap! *register-graph-updates-sent assoc-in [ws 0] false)
                   (throw e)))]
@@ -98,15 +141,8 @@
                                     max-remote-schema-version db-schema/version major-schema-version)
                          :repo repo
                          :graph-uuid graph-uuid
-                         :remote-schema-version max-remote-schema-version})))
-        (let [t (client-op/get-local-tx repo)]
-          (when (or (nil? @*last-calibrate-t)
-                    (< 500 (- t @*last-calibrate-t)))
-            (let [{:keys [server-schema-version _server-builtin-db-idents]}
-                  (m/? (r.skeleton/new-task--calibrate-graph-skeleton
-                        get-ws-create-task graph-uuid major-schema-version @conn))]
-              (reset! *server-schema-version server-schema-version))
-            (reset! *last-calibrate-t t))))
+                         :remote-schema-version max-remote-schema-version}))
+          (apply-remote-updates-from-apply-ops init-request-resp graph-uuid repo conn date-formatter add-log-fn)))
       ws)))
 
 (defn- ->pos
@@ -464,14 +500,4 @@
                    :ops [] :t-before (or local-tx 1)}
           r (m/? (ws-util/send&recv get-ws-create-task message))]
       (r.throttle/add-rtc-api-call-record! message)
-      (if-let [remote-ex (:ex-data r)]
-        (do (add-log-fn :rtc.log/pull-remote-data (assoc remote-ex :sub-type :pull-remote-data-exception))
-            (case (:type remote-ex)
-              :graph-lock-failed nil
-              :graph-lock-missing (throw r.ex/ex-remote-graph-lock-missing)
-              :rtc.exception/get-s3-object-failed nil
-              ;;else
-              (throw (ex-info "Unavailable3" {:remote-ex remote-ex}))))
-        (do (assert (pos? (:t r)) r)
-            (r.remote-update/apply-remote-update
-             graph-uuid repo conn date-formatter {:type :remote-update :value r} add-log-fn))))))
+      (apply-remote-updates-from-apply-ops r graph-uuid repo conn date-formatter add-log-fn))))

src/main/frontend/worker/rtc/core.cljs

@@ -209,8 +209,8 @@
         {:keys [*current-ws get-ws-create-task]}
         (gen-get-ws-create-map--memoized ws-url)
         get-ws-create-task (r.client/ensure-register-graph-updates--memoized
-                            get-ws-create-task graph-uuid major-schema-version
-                            repo conn *last-calibrate-t *online-users *server-schema-version add-log-fn)
+                            get-ws-create-task graph-uuid major-schema-version repo conn date-formatter
+                            *last-calibrate-t *online-users *server-schema-version add-log-fn)
         {:keys [assets-sync-loop-task]}
         (r.asset/create-assets-sync-loop repo get-ws-create-task graph-uuid major-schema-version conn *auto-push?)
         mixed-flow                 (create-mixed-flow repo get-ws-create-task *auto-push? *online-users)]
@@ -226,9 +226,9 @@
       (m/sp
         (try
           (log/info :rtc :loop-starting)
+          (started-dfv true)
           ;; init run to open a ws
           (m/? get-ws-create-task)
-          (started-dfv true)
           (update-remote-schema-version! conn @*server-schema-version)
           (reset! *assets-sync-loop-canceler
                   (c.m/run-task :assets-sync-loop-task

src/main/frontend/worker/rtc/malli_schema.cljs

@@ -197,6 +197,12 @@
       [:map
        [:ex-message :string]
        [:ex-data [:map [:type :keyword]]]]]
+     ["init-request"
+      (concat [:map
+               [:max-remote-schema-version {:optional true} :string]
+               [:server-schema-version {:optional true} :string]
+               [:server-builtin-db-idents {:optional true} [:set :keyword]]]
+              (m/children apply-ops-response-schema))]
      ["register-graph-updates"
       [:map
        [:t :int]
@@ -252,6 +258,12 @@
      [:multi {:dispatch :action}
       ["list-graphs"
        [:map]]
+      ["init-request"
+       [:map
+        [:graph-uuid :uuid]
+        [:schema-version db-schema/major-schema-version-string-schema]
+        [:t-before :int]
+        [:get-graph-skeleton :boolean]]]
       ["register-graph-updates"
        [:map
         [:graph-uuid :uuid]

src/main/frontend/worker/rtc/skeleton.cljs

@@ -2,11 +2,9 @@
   "Validate skeleton data between server and client"
   (:require [clojure.data :as data]
             [datascript.core :as d]
-            [frontend.worker.rtc.ws-util :as ws-util]
             [lambdaisland.glogi :as log]
             [logseq.db :as ldb]
-            [logseq.db.frontend.schema :as db-schema]
-            [missionary.core :as m]))
+            [logseq.db.frontend.schema :as db-schema]))
 
 (defn- get-builtin-db-idents
   [db]
@@ -18,29 +16,15 @@
          [?b :logseq.property/built-in?]]
        db))
 
-(defn new-task--calibrate-graph-skeleton
-  [get-ws-create-task graph-uuid major-schema-version db]
-  (m/sp
-    (let [r (m/? (ws-util/send&recv get-ws-create-task
-                                    {:action "get-graph-skeleton"
-                                     :graph-uuid graph-uuid
-                                     :schema-version (str major-schema-version)}))]
-      (if-let [remote-ex (:ex-data r)]
-        (case (:type remote-ex)
-          :graph-lock-failed
-          (throw (ex-info "retry calibrate-graph-skeleton" {:missionary/retry true}))
-          ;; else
-          (do (log/info :remote-ex remote-ex)
-              (throw (ex-info "Unavailable2" {:remote-ex remote-ex}))))
-        (let [{:keys [server-schema-version server-builtin-db-idents]} r
-              client-builtin-db-idents (set (get-builtin-db-idents db))
-              client-schema-version (ldb/get-graph-schema-version db)]
-          (when-not (zero? (db-schema/compare-schema-version client-schema-version server-schema-version))
-            (log/warn "RTC schema error: client version doesn't match server's version"
-                      [client-schema-version server-schema-version]))
-          (let [[client-only server-only _]
-                (data/diff client-builtin-db-idents server-builtin-db-idents)]
-            (when (or (seq client-only) (seq server-only))
-              (log/warn :db-idents-diff {:client-only client-only
-                                         :server-only server-only})))
-          r)))))
+(defn calibrate-graph-skeleton
+  [server-schema-version server-builtin-db-idents db]
+  (let [client-builtin-db-idents (set (get-builtin-db-idents db))
+        client-schema-version (ldb/get-graph-schema-version db)]
+    (when-not (zero? (db-schema/compare-schema-version client-schema-version server-schema-version))
+      (log/warn "RTC schema error: client version doesn't match server's version"
+                [client-schema-version server-schema-version]))
+    (let [[client-only server-only _]
+          (data/diff client-builtin-db-idents server-builtin-db-idents)]
+      (when (or (seq client-only) (seq server-only))
+        (log/warn :db-idents-diff {:client-only client-only
+                                   :server-only server-only})))))