dasemu/Synapse
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fase 6: close the subs-loader in-flight dedup gap, raise batch limits, add /v1/invoke-batch

Browse Source 
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/<invoke-db-worker
call funnels through for a remote client) issued within a 10ms window
into one POST /v1/invoke-batch, generalizing subs-loader's own
same-method batching to every thread-api call a remote client makes.
Two invokes only land in the same batch when neither is awaited before
the other fires, so this can't reorder anything a caller depends on
sequentially.

New db_worker_node_test.cljs coverage: invoke-batch mixing success and
repo-mismatch failure in one request, and invoke-batch auth/method
enforcement.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Darío Sevilla
2026-08-26 11:13:04 +02:00
parent 902bf7f568
commit 2164bbdf1e
7 changed files with 335 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
src/main/frontend/db/subs.cljs
View File

@@ -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 <load-block [graph-id block-uuid]
(loader/load! graph-id [:block block-uuid] schedule-load-batch!))

38
src/main/frontend/db/subs_loader.cljs
View File

@@ -3,20 +3,22 @@
(:require [frontend.state :as state]
[promesa.core :as p]))
(def ^:private limits {:blocks 1000 :children 25 :resources 25})
(def ^:private limits {:blocks 1000 :children 200 :resources 200})
(def ^:private slot-kind {:block :blocks :children :children :resource :resources})
(defonce ^:private *batch (atom {}))
(defonce ^:private *inflight (atom {}))
(defonce ^:private *stats (atom {:dedup-hits 0 :new-entries 0 :flushes 0 :batch-sizes {}}))
(defn get-stats
"Snapshot of subs-loader batching counters: :dedup-hits counts load! calls
that reused an already-pending slot request within the same unflushed
batch window (see `load!`); :batch-sizes tallies entries per slot kind
across flushes. Dev-only instrumentation for the server-client migration
latency baseline (see docs/agent-guide); does not affect batching or the
`limits` above."
that reused an already-pending OR already-in-flight slot request (see
`load!` and `*inflight`) instead of firing a duplicate; :batch-sizes
tallies entries per slot kind across flushes. Dev-only instrumentation for
the server-client migration latency baseline (see docs/agent-guide); does
not affect batching or the `limits` above."
[]
@*stats)
@@ -57,10 +59,25 @@
(throw (ex-info "Missing renderer snapshot group"
{:slot-key slot-key}))))
(defn- untrack-inflight!
[entries]
(swap! *inflight
(fn [inflight]
(reduce (fn [m {:keys [graph-id slot-key]}]
(dissoc m [graph-id slot-key]))
inflight
entries))))
(defn- flush!
[]
(when @state/db-worker-ready?
(let [batch (take-batch!)]
;; Moved into *inflight* (not just dropped after take-batch! clears
;; *batch*) so a load! for the same slot arriving while this request is
;; still in the air reuses this promise instead of firing a duplicate
;; request -- *batch* alone only dedupes calls that land before the
;; batch is sent, not across the flush boundary (Fase 6).
(swap! *inflight into (map (fn [entry] [[(:graph-id entry) (:slot-key entry)] entry]) batch))
(swap! *stats
(fn [s]
(-> 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)))

63
src/main/frontend/persist_db/remote.cljs
View File

@@ -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)

38
src/main/frontend/worker/db_worker_node.cljs
View File

@@ -448,6 +448,34 @@
[^js buf]
(str "\"" (-> (crypto/createHash "sha1") (.update buf) (.digest "hex")) "\""))
(defn- <process-batch-invoke!
"One entry of a /v1/invoke-batch request, run through the same
repo-check/lock/invoke path as /v1/invoke but never rejecting -- a failed
item becomes {:ok false :error ...} in its slot of the batch response
instead of failing every other invoke that happened to coalesce into the
same HTTP request (Fase 6)."
[proxy bound-repo {:keys [method argsTransit args]}]
(let [method-kw (normalize-method-kw method)
method-str (normalize-method-str method)
args' (or argsTransit args)]
(-> (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 (-> (<invoke! proxy method-str method-kw args')
(p/finally (fn [] (invoke-finished! bound-repo))))]
{:ok true :resultTransit result})))
(p/catch (fn [error]
(let [data (ex-data error)]
{:ok false
:error {:code (invoke-error-code data)
:message (invoke-error-message error data)}}))))))
(defn- ws-upgrade-handler
[^js wss bound-repo port extra-allowed-hosts]
(fn [^js req ^js socket head]
@@ -611,6 +639,16 @@
(log-invoke-error! res error :thread-api/asset-write-bytes))))
(send-json! res 400 {:ok false :error {:code :invalid-request :message "invalid asset path"}}))
(= request-path "/v1/invoke-batch")
(if (= method "POST")
(-> (p/let [body (<read-body req)
{:keys [invokes]} (js->clj (js/JSON.parse body) :keywordize-keys true)
results (p/all (map #(<process-batch-invoke! proxy bound-repo %) invokes))]
(send-json! res 200 {:ok true :results results}))
(p/catch (fn [error]
(log-invoke-error! res error nil))))
(send-text! res 405 "method-not-allowed"))
(= request-path "/v1/invoke")
(if (= method "POST")
(->

120
src/test/frontend/db/subs_loader_test.cljs Normal file
View File

@@ -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/<invoke-db-worker wraps the actual worker call in a p/let, so even
its first binding is deferred to a microtask rather than run synchronously
-- every step below that depends on the mock worker having been called
needs an explicit (p/delay 0) yield first.
The mock worker below only tracks :thread-api/get-render-snapshots calls --
frontend.background-tasks registers its own state/db-worker-ready? watchers
(online-status/search-idle-status sync) that fire real state/*db-worker
calls as an unrelated side effect of resetting it here, and this test has
no business asserting on those."
(:require [cljs.test :refer [async deftest is use-fixtures]]
[frontend.db.subs-loader :as loader]
[frontend.state :as state]
[promesa.core :as p]))
(def ^:private previous-worker (atom nil))
(use-fixtures :each
{:before #(reset! previous-worker @state/*db-worker)
:after #(reset! state/*db-worker @previous-worker)})
(defn- finish-async!
[done promise]
(-> 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 {}}))))))

10
src/test/frontend/db/subs_test.cljs
View File

@@ -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))))

75
src/test/frontend/worker/db_worker_node_test.cljs
View File

@@ -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)