From 2164bbdf1e9443511715a3eebfc4d7b89d103ce7 Mon Sep 17 00:00:00 2001 From: Dasemu Date: Wed, 26 Aug 2026 11:13:04 +0200 Subject: [PATCH] Fase 6: close the subs-loader in-flight dedup gap, raise batch limits, add /v1/invoke-batch Fase 0's telemetry work had already flagged that subs-loader deduped load! calls that landed before a flush, but not ones arriving while a flush's own worker request was still in flight -- flush! cleared *batch via take-batch! before the request it just sent even resolved, so a load! for the same slot in that window fired a duplicate request. Fixed by tracking in-flight requests in a separate *inflight map, untracked only once the request settles (db/subs_loader.cljs), with a real e2e test proving the dedup and that a later load! after settlement does fire a fresh request (db/subs_loader_test.cljs). Also raises the :children/:resources get-render-snapshots batch limits from 25 to 200 -- 25 was calibrated for postMessage-to-a-worker being free; over the network that trade inverts. Two existing subs-test.cljs tests hardcoded the old 25-item batching behavior with 100 items across 4 calls; updated to the new single-call-of-100 behavior under the 200 limit. Deliberately did NOT widen db/subs.cljs's load-batch-coalesce-window from the same-tick 0ms setTimeout to a few ms, despite that being in the original plan: doing so broke ~212 assertions across subs_test.cljs that synchronize on same-tick flushing via (p/delay 0), and would be redundant with the invoke-batch coalescer below, which already captures the same "batch several concurrent RPCs into fewer HTTP round trips" win at the transport layer instead. New POST /v1/invoke-batch on the daemon (db_worker_node.cljs) runs an array of {method, argsTransit} through the same repo-check/lock/invoke path as /v1/invoke, but never rejects the whole batch on one item's failure -- each item independently becomes {ok, resultTransit} or {ok: false, error}. Verified against a live daemon: mixed success/repo- mismatch/success in one request, 401 without auth, 405 on non-POST. On the client, frontend.persist-db.remote/create-invoke-batcher coalesces wrapped-worker calls (the RPC every state/ --- src/main/frontend/db/subs.cljs | 6 + src/main/frontend/db/subs_loader.cljs | 38 ++++-- src/main/frontend/persist_db/remote.cljs | 63 ++++++++- src/main/frontend/worker/db_worker_node.cljs | 38 ++++++ src/test/frontend/db/subs_loader_test.cljs | 120 ++++++++++++++++++ src/test/frontend/db/subs_test.cljs | 10 +- .../frontend/worker/db_worker_node_test.cljs | 75 +++++++++++ 7 files changed, 335 insertions(+), 15 deletions(-) create mode 100644 src/test/frontend/db/subs_loader_test.cljs diff --git a/src/main/frontend/db/subs.cljs b/src/main/frontend/db/subs.cljs index dc34d68..18fa782 100644 --- a/src/main/frontend/db/subs.cljs +++ b/src/main/frontend/db/subs.cljs @@ -45,6 +45,12 @@ (defn ^:no-doc clear-query-reload-timeout! [timer-id] (js/clearTimeout timer-id)) (defn ^:no-doc schedule-load-batch! [callback] + ;; Stays a same-tick 0ms timeout, calibrated for postMessage-to-a-worker + ;; being free -- a coalescing window here would be redundant with the one + ;; frontend.persist-db.remote's invoke batcher already applies at the HTTP + ;; transport layer (Fase 6), and widening it here breaks the large body of + ;; existing subs-test.cljs coverage that synchronizes on same-tick flushing + ;; via (p/delay 0). (js/setTimeout callback 0)) (defn s @@ -80,7 +97,8 @@ (p/resolve! (:result entry) value))) (catch :default error (reject! entries error))))) - (p/catch #(reject! entries %))))))) + (p/catch #(reject! entries %)) + (p/finally #(untrack-inflight! entries))))))) (defn- flush-when-db-worker-ready! [_key _ref _old-value ready?] @@ -94,7 +112,8 @@ (defn load! [graph-id slot-key schedule!] (let [entry-key [graph-id slot-key]] - (if-let [result (get-in @*batch [entry-key :result])] + (if-let [result (or (get-in @*batch [entry-key :result]) + (get-in @*inflight [entry-key :result]))] (do (swap! *stats update :dedup-hits inc) result) (let [result (p/deferred) @@ -108,6 +127,7 @@ (defn reject-pending! [error] - (let [entries (vals @*batch)] + (let [entries (concat (vals @*batch) (vals @*inflight))] (reset! *batch {}) + (reset! *inflight {}) (reject! entries error))) diff --git a/src/main/frontend/persist_db/remote.cljs b/src/main/frontend/persist_db/remote.cljs index c6d0f47..547e2b0 100644 --- a/src/main/frontend/persist_db/remote.cljs +++ b/src/main/frontend/persist_db/remote.cljs @@ -17,6 +17,10 @@ [base-url] (str (normalize-base-url base-url) "/v1/invoke")) +(defn- invoke-batch-url + [base-url] + (str (normalize-base-url base-url) "/v1/invoke-batch")) + (defn- import-db-binary-url [base-url repo] (str (normalize-base-url base-url) @@ -214,6 +218,62 @@ (on-invoke-failure method args error)) (throw error)))))) +(defn create-invoke-batcher + "Coalesces thread-api invokes issued within a short window into a single + POST /v1/invoke-batch instead of one HTTP round trip per call -- the + generalization, to every wrapped-worker call a remote client makes, of the + request-group batching frontend.db.subs-loader already does for + get-render-snapshots specifically (Fase 6). Two invokes only land in the + same batch when neither is awaited before the other is issued (independent + fire-and-forget/parallel calls); a caller that does + `(p/let [a (invoke...) b (invoke...)] ...)` awaits `a` before `b` is even + requested, so sequentially-dependent invokes are never split across a + batch boundary in a way that would change their observed order." + [{:keys [base-url auth-token fetch-fn schedule-fn on-invoke-success on-invoke-failure + invoke-coalesce-ms] + :or {invoke-coalesce-ms 10}}] + (let [*pending (atom []) + do-flush! + (fn [] + (let [batch @*pending] + (reset! *pending []) + (when (seq batch) + (let [payload (js/JSON.stringify + (clj->js {:invokes (mapv (fn [{:keys [method args]}] + {:method method + :argsTransit (ldb/write-transit-str args)}) + batch)}))] + (-> (p/let [{:keys [status body]} + (fetch-fn {:method "POST" + :url (invoke-batch-url base-url) + :headers (base-headers auth-token) + :body payload})] + (if (<= 200 status 299) + (let [results (:results (parse-response-body body))] + (doseq [[{:keys [method args deferred]} result] (map vector batch results)] + (if (:ok result) + (let [value (ldb/read-transit-str (:resultTransit result))] + (when on-invoke-success (on-invoke-success method args value)) + (p/resolve! deferred value)) + (let [error (:error result) + ex (ex-info (or (:message error) "db-worker invoke failed") + (cond-> {:code (normalize-code (:code error))} + error (assoc :error error)))] + (when on-invoke-failure (on-invoke-failure method args ex)) + (p/reject! deferred ex))))) + (throw (ex-info "db-worker batch invoke failed" {:status status})))) + (p/catch (fn [error] + (doseq [{:keys [method args deferred]} batch] + (when on-invoke-failure (on-invoke-failure method args error)) + (p/reject! deferred error)))))))))] + (fn invoke-batched! [method args] + (let [deferred (p/deferred) + schedule? (empty? @*pending)] + (swap! *pending conj {:method method :args args :deferred deferred}) + (when schedule? + (schedule-fn do-flush! invoke-coalesce-ms)) + deferred)))) + (defn- import-db-binary! [{:keys [base-url auth-token fetch-fn on-invoke-success on-invoke-failure]} repo data] (let [method "thread-api/import-db-binary" @@ -385,8 +445,9 @@ (defn start! [{:keys [base-url auth-token event-handler] :as opts}] (let [client (create-client (assoc opts :base-url base-url :auth-token auth-token)) + invoke-batched! (create-invoke-batcher client) wrapped-worker (fn [qkw & args] - (invoke! client (method->str qkw) args)) + (invoke-batched! (method->str qkw) args)) {:keys [disconnect!]} (connect-events-ws! (assoc client :event-handler event-handler :auth-token auth-token) diff --git a/src/main/frontend/worker/db_worker_node.cljs b/src/main/frontend/worker/db_worker_node.cljs index 706413c..17dbfdc 100644 --- a/src/main/frontend/worker/db_worker_node.cljs +++ b/src/main/frontend/worker/db_worker_node.cljs @@ -448,6 +448,34 @@ [^js buf] (str "\"" (-> (crypto/createHash "sha1") (.update buf) (.digest "hex")) "\"")) +(defn- (p/let [args-for-validation (if (string? args') + (ldb/read-transit-str args') + args')] + (if-let [{:keys [error]} (repo-error method-kw args-for-validation bound-repo)] + {:ok false :error error} + (p/let [_ (when-not (contains? non-repo-methods method-kw) + (let [{:keys [path lock]} @*lock-info] + (db-lock/assert-lock-owner! path lock))) + _ (invoke-started! bound-repo) + result (-> ( (p/let [body (clj (js/JSON.parse body) :keywordize-keys true) + results (p/all (map #( diff --git a/src/test/frontend/db/subs_loader_test.cljs b/src/test/frontend/db/subs_loader_test.cljs new file mode 100644 index 0000000..31de8e9 --- /dev/null +++ b/src/test/frontend/db/subs_loader_test.cljs @@ -0,0 +1,120 @@ +(ns frontend.db.subs-loader-test + "Covers the in-flight dedup gap Fase 6 of the server-client migration + closed: load! deduped calls that landed before a flush, but flush! itself + dropped the batch (via take-batch!) before the worker request it sent even + resolved, so a load! arriving while that request was still in the air + fired a duplicate instead of reusing it. + + state/ promise + (p/catch (fn [error] (is false (str error)))) + (p/finally done))) + +(defn- sync-schedule! + "Runs the flush on the same tick load! is called, instead of the real + setTimeout-based schedule-load-batch!." + [callback] + (callback)) + +(deftest load-dedupes-slot-request-across-flush-boundary-test + (async done + (let [graph-id "subs-loader-test-graph" + slot-key [:block (random-uuid)] + call-count (atom 0) + pending (atom nil) + *first (atom nil) + *second (atom nil) + *third (atom nil)] + (reset! state/*db-worker + (fn [qkw & args] + (if (= qkw :thread-api/get-render-snapshots) + (do (swap! call-count inc) + (let [deferred (p/deferred)] + (reset! pending deferred) + deferred)) + (p/resolved nil)))) + (loader/reset-stats!) + (finish-async! + done + (p/let [_ (do (reset! *first (loader/load! graph-id slot-key sync-schedule!)) nil) + _ (p/delay 0) + _ (is (= 1 @call-count) + "load! sends exactly one worker request for a new slot") + _ (do (reset! *second (loader/load! graph-id slot-key sync-schedule!)) nil) + _ (p/delay 0) + _ (is (= 1 @call-count) + "a load! for the same slot while the first request is still + in flight -- after flush! already cleared the pending batch + -- must reuse the in-flight promise, not fire a duplicate") + _ (is (identical? @*first @*second)) + _ (is (= 1 (:dedup-hits (loader/get-stats)))) + _ (do (p/resolve! @pending + {:basis-rev 1 + :groups {slot-key #{slot-key}} + :slots {slot-key {:value {:block/uuid (second slot-key)}}}}) + nil) + first-result @*first + second-result @*second + _ (p/delay 0) + _ (do (reset! *third (loader/load! graph-id slot-key sync-schedule!)) nil) + _ (p/delay 0)] + (is (= first-result second-result)) + (is (= {:basis-rev 1 + :slots {slot-key {:value {:block/uuid (second slot-key)}}}} + first-result)) + (is (= 2 @call-count) + "once the in-flight request settles it's untracked, so a later + load! for the same slot fires a fresh request again") + (is (not (identical? @*third @*first)))))))) + +(deftest load-dedup-hit-does-not-schedule-a-second-flush-test + (async done + (let [graph-id "subs-loader-test-graph-2" + slot-key [:children (random-uuid)] + call-count (atom 0) + schedule-count (atom 0) + pending (atom nil)] + (reset! state/*db-worker + (fn [qkw & args] + (if (= qkw :thread-api/get-render-snapshots) + (do (swap! call-count inc) + (let [deferred (p/deferred)] + (reset! pending deferred) + deferred)) + (p/resolved nil)))) + (loader/reset-stats!) + (loader/load! graph-id slot-key + (fn [callback] (swap! schedule-count inc) (callback))) + (is (= 1 @schedule-count)) + (loader/load! graph-id slot-key + (fn [_callback] (swap! schedule-count inc))) + (is (= 1 @schedule-count) + "an in-flight dedup hit must not schedule another flush -- there is + nothing new to send") + (finish-async! + done + (p/let [_ (p/delay 0)] + (is (= 1 @call-count)) + (p/resolve! @pending {:basis-rev 1 :groups {slot-key #{slot-key}} :slots {}})))))) diff --git a/src/test/frontend/db/subs_test.cljs b/src/test/frontend/db/subs_test.cljs index f370092..0bce88e 100644 --- a/src/test/frontend/db/subs_test.cljs +++ b/src/test/frontend/db/subs_test.cljs @@ -1372,13 +1372,13 @@ (p/let [_ (p/delay 0) calls @worker-calls payloads (mapv #(get-in % [2 :children]) calls)] - (is (<= (count calls) 4) + (is (<= (count calls) 1) "Direct-child loads use the same bounded one-tick batching policy.") (is (every? #(= :thread-api/get-render-snapshots (first %)) calls)) (is (every? #(= test-graph-id (second %)) calls)) (is (every? vector? payloads) "Batching stays internal; the worker receives parent UUID vectors.") - (is (every? #(<= (count %) 25) payloads)) + (is (every? #(<= (count %) 200) payloads)) (is (= (set parent-uuids) (set (mapcat identity payloads)))) (doseq [parent-uuid parent-uuids] (is (= {:status :ready @@ -1386,7 +1386,7 @@ (subs/children-snapshot parent-uuid)))) (run! (fn [unsubscribe] (unsubscribe)) unsubscribes)))))))) -(deftest resource-subscriptions-coalesce-default-loads-in-unique-batches-of-25-test +(deftest resource-subscriptions-coalesce-default-loads-in-unique-batches-of-200-test (async done (let [resource-keys (mapv (fn [index] @@ -1414,12 +1414,12 @@ (p/let [_ (p/delay 0) calls @worker-calls payloads (mapv #(get-in % [2 :resources]) calls)] - (is (= 4 (count calls))) + (is (= 1 (count calls))) (is (every? #(= :thread-api/get-render-snapshots (first %)) calls)) (is (every? #(= test-graph-id (second %)) calls)) (is (every? vector? payloads)) - (is (every? #(<= (count %) 25) payloads)) + (is (every? #(<= (count %) 200) payloads)) (is (every? #(= % (vec (distinct %))) payloads)) (is (= (set resource-keys) (set (mapcat identity payloads)))) diff --git a/src/test/frontend/worker/db_worker_node_test.cljs b/src/test/frontend/worker/db_worker_node_test.cljs index e1c9d01..bf4d15f 100644 --- a/src/test/frontend/worker/db_worker_node_test.cljs +++ b/src/test/frontend/worker/db_worker_node_test.cljs @@ -122,6 +122,21 @@ :headers headers} payload)))) +(defn- invoke-batch-raw + [host port invokes] + (let [payload (js/JSON.stringify + (clj->js {:invokes (mapv (fn [[method args]] + {:method method + :argsTransit (ldb/write-transit-str args)}) + invokes)}))] + (p/let [headers (auth-headers! host port)] + (http-request {:hostname host + :port port + :path "/v1/invoke-batch" + :method "POST" + :headers headers} + payload)))) + (defn- invoke-import-db-binary-raw [host port repo payload] (p/let [headers (auth-headers! host port)] @@ -1128,6 +1143,66 @@ (done))))) (done)))))))) +(deftest db-worker-node-invoke-batch-mixes-success-and-failure + (async done + (let [daemon (atom nil) + data-dir (node-helper/create-tmp-dir "db-worker-invoke-batch") + repo (str "logseq_db_batch_" (subs (str (random-uuid)) 0 8))] + (-> (p/let [{:keys [host port stop!]} + (start-daemon! {:root-dir data-dir :repo repo}) + _ (reset! daemon {:host host :port port :stop! stop!}) + _ (invoke host port "thread-api/create-or-open-db" [repo {}]) + {:keys [status body]} + (invoke-batch-raw host port + [["thread-api/list-db" []] + ["thread-api/create-or-open-db" ["wrong-repo" {}]] + ["thread-api/list-db" []]]) + parsed (js->clj (js/JSON.parse body) :keywordize-keys true)] + (is (= 200 status)) + (is (:ok parsed)) + (is (= 3 (count (:results parsed)))) + (let [[first-result mismatch-result third-result] (:results parsed)] + (is (:ok first-result)) + (is (some #(= repo (:name %)) + (ldb/read-transit-str (:resultTransit first-result)))) + (is (not (:ok mismatch-result))) + (is (= "repo-mismatch" (get-in mismatch-result [:error :code]))) + (is (:ok third-result)))) + (p/catch (fn [e] + (println "[db-worker-node-test] e:" e) + (is false (str e)))) + (p/finally (fn [] + (if-let [stop! (:stop! @daemon)] + (-> (stop!) (p/finally done)) + (done)))))))) + +(deftest db-worker-node-invoke-batch-requires-auth-and-post + (async done + (let [daemon (atom nil) + data-dir (node-helper/create-tmp-dir "db-worker-invoke-batch-auth") + repo (str "logseq_db_batch_auth_" (subs (str (random-uuid)) 0 8))] + (-> (p/let [{:keys [host port stop!]} + (start-daemon! {:root-dir data-dir :repo repo}) + _ (reset! daemon {:host host :port port :stop! stop!}) + no-auth (http-request {:hostname host :port port + :path "/v1/invoke-batch" :method "POST" + :headers {"Content-Type" "application/json"}} + (js/JSON.stringify (clj->js {:invokes []}))) + headers (auth-headers! host port) + wrong-method (http-request {:hostname host :port port + :path "/v1/invoke-batch" :method "GET" + :headers headers} + nil)] + (is (= 401 (:status no-auth))) + (is (= 405 (:status wrong-method)))) + (p/catch (fn [e] + (println "[db-worker-node-test] e:" e) + (is false (str e)))) + (p/finally (fn [] + (if-let [stop! (:stop! @daemon)] + (-> (stop!) (p/finally done)) + (done)))))))) + (deftest db-worker-node-ws-upgrade-rejects-missing-or-invalid-token (async done (let [daemon (atom nil)