fix: switch graph if db-worker-node server unreachable

Author: RCmerci

docs/agent-guide/db-worker/002-desktop-db-worker-request-cap-switch-graph.md

@@ -225,6 +225,12 @@
   [window-id]
   (ensure-window-stopped! manager window-id))
 
+(defn release-runtime!
+  ([repo window-id]
+   (release-runtime! manager repo window-id))
+  ([mgr repo window-id]
+   (ensure-stopped! mgr repo window-id)))
+
 (defn release-repo!
   [repo]
   (ensure-repo-stopped! manager repo))

src/electron/electron/db_worker.cljs

@@ -234,6 +234,11 @@
     (p/rejected (ex-info "repo is required" {:code :missing-repo}))
     (db-worker/ensure-runtime! (canonical-repo repo) (.-id window))))
 
+(defmethod handle :releaseDbWorkerRuntime [^js window [_ repo]]
+  (if (string/blank? repo)
+    (p/rejected (ex-info "repo is required" {:code :missing-repo}))
+    (db-worker/release-runtime! (canonical-repo repo) (.-id window))))
+
 (defmethod handle :db-export [window [_ repo force-backup?]]
   (when-let [repo (canonical-repo repo)]
     (db/ensure-graph-dir! repo)

src/electron/electron/handler.cljs

@@ -0,0 +1,37 @@
+(ns frontend.handler.graph-failover
+  "Switches away from an unavailable active graph without deleting graph data."
+  (:require [frontend.context.i18n :refer [t]]
+            [frontend.handler.notification :as notification]
+            [frontend.state :as state]
+            [frontend.util.text :as text-util]))
+
+(defn- graph-name
+  [repo]
+  (text-util/get-graph-name-from-path repo))
+
+(defn- fallback-graph
+  [failed-repo]
+  (some (fn [{:keys [url]}]
+          (when (not= failed-repo url)
+            url))
+        (state/get-repos)))
+
+(defn switch-away-from-current-repo!
+  ([failed-repo]
+   (switch-away-from-current-repo! failed-repo {}))
+  ([failed-repo _opts]
+   (when (= failed-repo (state/get-current-repo))
+     (state/set-current-repo! nil)
+     (if-let [graph (fallback-graph failed-repo)]
+       (do
+         (notification/show! (t :graph.switch/db-worker-unavailable-switching-warning
+                                (graph-name failed-repo)
+                                (graph-name graph))
+                             :warning)
+         (state/pub-event! [:graph/switch graph {:persist? false}])
+         graph)
+       (do
+         (notification/show! (t :graph.switch/db-worker-unavailable-no-fallback-warning
+                                (graph-name failed-repo))
+                             :warning)
+         nil)))))

src/main/frontend/handler/graph_failover.cljs

@@ -3,6 +3,7 @@
   (:require [electron.ipc :as ipc]
             [frontend.config :as config]
             [frontend.db.transact :as db-transact]
+            [frontend.handler.graph-failover :as graph-failover]
             [frontend.persist-db.browser :as browser]
             [frontend.persist-db.protocol :as protocol]
             [frontend.persist-db.remote :as remote]
@@ -13,12 +14,16 @@
             [logseq.db :as ldb]
             [promesa.core :as p]))
 
+(def max-db-worker-request-failures 3)
+
 (defonce opfs-db (browser/->InBrowser))
 (defonce remote-db (atom nil))
 (defonce remote-repo (atom nil))
+(defonce remote-runtime-state (atom nil))
 
 (defn- clear-remote-runtime!
   []
+  (reset! remote-runtime-state nil)
   (reset! remote-db nil)
   (reset! remote-repo nil)
   (reset! state/*db-worker nil))
@@ -37,6 +42,78 @@
         (p/resolved true)))
     (p/resolved false)))
 
+(defn- set-remote-runtime!
+  [repo client session-id]
+  (reset! remote-runtime-state {:repo repo
+                                :client client
+                                :session-id session-id
+                                :request-failures 0
+                                :failover-triggered? false})
+  (reset! remote-db client)
+  (reset! remote-repo repo)
+  (reset! state/*db-worker (:wrapped-worker client)))
+
+(defn- active-runtime-session?
+  [state repo session-id]
+  (and (= repo (:repo state))
+       (= session-id (:session-id state))))
+
+(defn- reset-active-request-failures!
+  [repo session-id]
+  (swap! remote-runtime-state
+         (fn [state]
+           (if (active-runtime-session? state repo session-id)
+             (assoc state :request-failures 0)
+             state))))
+
+(defn- server-unavailable-error?
+  [error]
+  (let [{:keys [status code]} (ex-data error)]
+    (or (nil? (ex-data error))
+        (= :server-unavailable code)
+        (= :db-worker-unavailable code)
+        (= :connection-refused code)
+        (= :fetch-failed code)
+        (= :network-error code)
+        (= 0 status))))
+
+(defn- trigger-db-worker-failover!
+  [repo remote-client]
+  (when remote-client
+    (-> (remote/stop! remote-client)
+        (p/catch (fn [error]
+                   (log/warn :db-worker-failover-stop-error {:repo repo
+                                                             :error error})))))
+  (when (= repo @remote-repo)
+    (clear-remote-runtime!))
+  (-> (ipc/ipc "releaseDbWorkerRuntime" repo)
+      (p/catch (fn [error]
+                 (log/warn :db-worker-failover-release-runtime-error {:repo repo
+                                                                       :error error}))))
+  (graph-failover/switch-away-from-current-repo! repo {:reason :db-worker-request-failed}))
+
+(defn- record-active-request-failure!
+  [repo session-id error]
+  (when (server-unavailable-error? error)
+    (let [triggered? (atom false)
+          remote-client (atom nil)]
+      (swap! remote-runtime-state
+             (fn [state]
+               (if (and (active-runtime-session? state repo session-id)
+                        (not (:failover-triggered? state)))
+                 (let [failures (inc (or (:request-failures state) 0))]
+                   (if (>= failures max-db-worker-request-failures)
+                     (do
+                       (reset! triggered? true)
+                       (reset! remote-client (:client state))
+                       (assoc state
+                              :request-failures failures
+                              :failover-triggered? true))
+                     (assoc state :request-failures failures)))
+                 state)))
+      (when @triggered?
+        (trigger-db-worker-failover! repo @remote-client)))))
+
 (defn- node-runtime?
   []
   (and (exists? js/process)
@@ -57,26 +134,31 @@
   [repo]
   (if (or (nil? repo) (= repo @remote-repo))
     (p/resolved @remote-db)
-    (p/let [_ (when @remote-db
-                (remote/stop! @remote-db))
-            runtime (ipc/ipc "db-worker-runtime" repo)
-            client (remote/start! (assoc runtime
-                                         :repo repo
-                                         :event-handler worker-handler/handle))]
-      (reset! remote-db client)
-      (reset! remote-repo repo)
-      (reset! state/*db-worker (:wrapped-worker client))
-      (p/let [_ (state/<invoke-db-worker :thread-api/set-db-sync-config
-                                         (current-db-sync-config))]
-        nil)
-      (ldb/register-transact-fn!
-       (fn remote-transact!
-         [repo tx-data tx-meta]
-         (db-transact/transact browser/transact!
-                               (if (string? repo) repo (state/get-current-repo))
-                               tx-data
-                               (assoc tx-meta :client-id (:client-id @state/state)))))
-      client)))
+    (let [session-id (str (random-uuid))]
+      (p/let [_ (when @remote-db
+                  (remote/stop! @remote-db))
+              runtime (ipc/ipc "db-worker-runtime" repo)
+              client (remote/start! (assoc runtime
+                                           :repo repo
+                                           :event-handler worker-handler/handle
+                                           :on-invoke-success (fn [_method _args _result]
+                                                                (reset-active-request-failures! repo session-id))
+                                           :on-invoke-failure (fn [_method _args error]
+                                                                (record-active-request-failure! repo session-id error))
+                                           :on-event-error (fn [error]
+                                                             (record-active-request-failure! repo session-id error))))]
+        (set-remote-runtime! repo client session-id)
+        (p/let [_ (state/<invoke-db-worker :thread-api/set-db-sync-config
+                                           (current-db-sync-config))]
+          nil)
+        (ldb/register-transact-fn!
+         (fn remote-transact!
+           [repo tx-data tx-meta]
+           (db-transact/transact browser/transact!
+                                 (if (string? repo) repo (state/get-current-repo))
+                                 tx-data
+                                 (assoc tx-meta :client-id (:client-id @state/state)))))
+        client))))
 
 (defn <start-runtime!
   []

src/main/frontend/persist_db.cljs

@@ -87,37 +87,49 @@
               {:close! (fn []
                          (.close es))})
             {:close! (fn [] nil)}))]
-  (assoc opts
-         :fetch-fn (or fetch-fn default-fetch-fn)
-         :open-sse-fn (or open-sse-fn default-open-sse-fn)
-         :schedule-fn (or schedule-fn (fn [f delay-ms]
-                                        (js/setTimeout f delay-ms)))
-         :reconnect-delay-ms (or reconnect-delay-ms 1000))))
+    (assoc opts
+           :fetch-fn (or fetch-fn default-fetch-fn)
+           :open-sse-fn (or open-sse-fn default-open-sse-fn)
+           :schedule-fn (or schedule-fn (fn [f delay-ms]
+                                          (js/setTimeout f delay-ms)))
+           :reconnect-delay-ms (or reconnect-delay-ms 1000))))
 
 (defn invoke!
-  [{:keys [base-url auth-token fetch-fn]} method args]
+  [{:keys [base-url auth-token fetch-fn on-invoke-success on-invoke-failure]} method args]
   (let [payload (js/JSON.stringify
                  (clj->js {:method method
                            :argsTransit (ldb/write-transit-str args)}))]
-    (p/let [{:keys [status body]}
-            (fetch-fn {:method "POST"
-                       :url (invoke-url base-url)
-                       :headers (base-headers auth-token)
-                       :body payload})
-            parsed (parse-response-body body)]
-      (if (<= 200 status 299)
-        (ldb/read-transit-str (:resultTransit parsed))
-        (let [error (:error parsed)]
-          (throw (ex-info (or (:message error) "db-worker invoke failed")
-                          (cond-> {:status status
-                                   :code (normalize-code (:code error))}
-                            error (assoc :error error)))))))))
+    (->
+     (p/let [{:keys [status body]}
+             (fetch-fn {:method "POST"
+                        :url (invoke-url base-url)
+                        :headers (base-headers auth-token)
+                        :body payload})
+             parsed (parse-response-body body)]
+       (if (<= 200 status 299)
+         (let [result (ldb/read-transit-str (:resultTransit parsed))]
+           (when on-invoke-success
+             (on-invoke-success method args result))
+           result)
+         (let [error (:error parsed)]
+           (throw (ex-info (or (:message error) "db-worker invoke failed")
+                           (cond-> {:status status
+                                    :code (normalize-code (:code error))}
+                             error (assoc :error error)))))))
+     (p/catch (fn [error]
+                (when on-invoke-failure
+                  (on-invoke-failure method args error))
+                (throw error))))))
 
 (defn connect-events!
-  [{:keys [base-url auth-token event-handler open-sse-fn schedule-fn reconnect-delay-ms]} wrapped-worker]
+  [{:keys [base-url auth-token event-handler open-sse-fn schedule-fn reconnect-delay-ms on-event-error]} wrapped-worker]
   (let [connected? (atom true)
         buffer (atom "")
         subscription (atom nil)
+        close-subscription! (fn []
+                              (when-let [close! (:close! @subscription)]
+                                (close!))
+                              (reset! subscription nil))
         dispatch! (fn [event-str]
                     (when-let [line (data-line event-str)]
                       (let [event (parse-response-body line)
@@ -140,14 +152,17 @@
                                                 (reset! buffer next-buffer)
                                                 (dispatch! event-str)
                                                 (recur))))))
-                          :on-error (fn [_error]
+                          :on-error (fn [error]
                                       (when @connected?
-                                        (schedule-fn open! reconnect-delay-ms)))}))))]
+                                        (close-subscription!)
+                                        (when on-event-error
+                                          (on-event-error error))
+                                        (when @connected?
+                                          (schedule-fn open! reconnect-delay-ms))))}))))]
       (open!)
       {:disconnect! (fn []
                       (reset! connected? false)
-                      (when-let [close! (:close! @subscription)]
-                        (close!))
+                      (close-subscription!)
                       nil)})))
 
 (defn- method->str

src/main/frontend/persist_db/remote.cljs

@@ -734,6 +734,8 @@
  :graph.page/title "Page graph"
 
  :graph.switch/add-graph-action "Yes, add another graph"
+ :graph.switch/db-worker-unavailable-no-fallback-warning "The database worker for graph \"{1}\" is unavailable, and no other graph is available."
+ :graph.switch/db-worker-unavailable-switching-warning "The database worker for graph \"{1}\" is unavailable. Switching to graph \"{2}\"."
  :graph.switch/empty-desc "No matched graphs. Do you want to add another one?"
  :graph.switch/prompt "Switch to:"
  :graph.switch/select-prompt "Select a graph"

src/resources/dicts/en.edn

@@ -731,6 +731,8 @@
  :graph.page/title "页面图谱"
 
  :graph.switch/add-graph-action "再添加一份图谱"
+ :graph.switch/db-worker-unavailable-no-fallback-warning "图谱“{1}”的数据库 worker 不可用，且没有其他可用图谱。"
+ :graph.switch/db-worker-unavailable-switching-warning "图谱“{1}”的数据库 worker 不可用。正在切换到图谱“{2}”。"
  :graph.switch/empty-desc "没有匹配的图谱。你想再添加一份图谱吗？"
  :graph.switch/prompt "切换到："
  :graph.switch/select-prompt "请选择一份图谱"

src/resources/dicts/zh-cn.edn

@@ -78,6 +78,26 @@
                      (is false (str "unexpected error: " e))))
           (p/finally (fn [] (done)))))))
 
+(deftest release-runtime-detaches-only-requested-window-repo-association
+  (async done
+    (let [stop-calls (atom [])
+          manager (db-worker/create-manager
+                   {:start-daemon! (fn [repo] (p/resolved (runtime repo)))
+                    :stop-daemon! (fn [rt]
+                                    (swap! stop-calls conj (:repo rt))
+                                    (p/resolved true))})]
+      (-> (p/let [_ (db-worker/ensure-started! manager "graph-a" :window-1)
+                  _ (db-worker/ensure-started! manager "graph-a" :window-2)
+                  _ (db-worker/release-runtime! manager "graph-a" :window-1)
+                  state @(:state manager)]
+            (is (empty? @stop-calls))
+            (is (nil? (get-in state [:window->repo :window-1])))
+            (is (= "graph-a" (get-in state [:window->repo :window-2])))
+            (is (= #{:window-2} (get-in state [:repos "graph-a" :windows]))))
+          (p/catch (fn [e]
+                     (is false (str "unexpected error: " e))))
+          (p/finally (fn [] (done)))))))
+
 (deftest ensure-stopped-stale-repo-does-not-clear-new-window-mapping
   (async done
     (let [stop-calls (atom [])