github/penpot
1
0
Fork 0
mirror of https://github.com/penpot/penpot.git synced 2026-10-29 05:50:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

🐛 Keep concurrency limits from deadlocking on nested calls

Browse Source 
An instance left the audit events pipeline dead: every push-audit-events
answered with 429 and 69 worker threads parked for good inside a limiter.

Two code paths of one profile acquired the same two limiters in opposite
order: a command declaring its own limits (upload-file-media-object) took
process-image/by-profile and then root/by-profile, while a command with no
limits of its own took the roots first and then re-entered climit from
inside its handler with the same ids (create-file-media-object-from-url).
Each path held what the other one waited for, and no limiter had a timeout
to break the wait.

Now wrap puts the root limiters after the command's own, so they end up
outermost and every path meets them first: the nested chains carry only the
command's own limits and run inside them, which puts both paths in the same
order. impl-invoke also keeps, per thread, the set of limiters that thread
is inside of and runs the handler inline when a call asks for one it
already holds, so a handler can no longer queue behind itself.

The shipped climit.edn sets a timeout on every limiter (60s for operations
that can be slow, 30s otherwise), so no wait can park a thread for ever.
The production file lives outside the repository and has to be mirrored.

Tests cover the acquisition order, the deadlock, the reentrancy and the
timeouts of the shipped config.

AI-assisted-by: deepseek-v4.1-flash
This commit is contained in:
Andrey Antukh 2026-10-12 12:04:31 +00:00
parent 18705f2d00
commit f242ff9574
8 changed files with 436 additions and 69 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
.serena/memories/backend/rpc-db-worker-subtleties.md
View File

@ -9,8 +9,11 @@
- Params with `::sm/params` are decoded/conformed through the JSON transformer and successful IObj results get `:encode/json` metadata. Legacy spec conforming only applies when no Malli params schema exists.
- Nil RPC bodies become HTTP 204 unless explicit status metadata is present. Stream bodies default to `application/octet-stream` when no content type is set.
- `::climit/id` is either a plain keyword, which gives one shared limiter per process, or the vector form `[[id key-fn]]`, which gives one limiter per key. `climit/wrap` takes the key only from that positional `key-fn` (`climit/get-limits` maps a plain keyword to a single-element vector), so `::climit/key-fn` is metadata that nothing reads: a command declaring `::climit/id :x/by-profile` next to it ends up with a single shared limiter, and the missing key is visible in the `app.rpc.climit` log lines, whose `id` carries no key suffix. That is what happened to `push-audit-events`, the last user of the dead metadata, from `a5c6d78ee5` (2024-01-27, "fix some fundamental bugs on climit module", which moved the key to the vector form without updating it) until it was fixed; nothing declares `::climit/key-fn` any more, so keep using the vector form for anything that must be per profile.
- The `penpot_rpc_climit_queue`/`permits` gauges carry only the limiter name, never the key, so every per-profile limiter of the same name collapses into one series and the last call wins: a per-key pile-up is invisible there and a healthy looking `cola=0` proves nothing. A stuck limiter does show up in `penpot_rpc_climit_timing_count` (the acquired path), which is a counter: compare the same limiter name across instances and the stuck one reads 0 acquisitions while its peer reads thousands.
- Nothing in the limiter path has a timeout unless the config sets one, and `psm/acquire` then calls `Semaphore.acquire`, which parks the calling thread forever. That is what turns a wait into a lost worker thread: a permanently parked thread still counts as in-system for every outer limiter of its chain, so the outermost one (the method's own by-profile limiter) stops granting permits and answers 429 to everything. A limiter without `:queue` in the config gets an unlimited queue and therefore never rejects: its pile-up is silent, with no 429 and no log line except the WARN `enqueued`/`acquired` lines of the keyed id, and its per-key nature is invisible in the metrics. Note that a request whose profile limit is exhausted parks *inside* the outer limiter while holding its permit, and handlers that assign `::climit/id` dynamically (media, profile) re-acquire the same keyed limiters, so a chain of such waits can close into a cycle that no timeout ever breaks.
- The `penpot_rpc_climit_queue`/`permits` gauges carry only the limiter name, never the key, so every per-profile limiter of the same name collapses into one series and the last call wins: a per-key pile-up is invisible there and a healthy looking `cola=0` proves nothing. A stuck limiter does show up in `penpot_rpc_climit_timing_count` (the acquired path), which is a counter: compare the same limiter name across instances and the stuck one reads 0 acquisitions while its peer reads thousands. In thread dumps it reads as threads parked in `psm/acquire` whose CPU time does not grow between two dumps, on an instance that is otherwise idle.
- Nothing in the limiter path has a timeout unless the config sets one, and `psm/acquire` then calls `Semaphore.acquire`, which parks the calling thread forever. A parked thread still counts as in-system for every outer limiter of its chain, so the outermost one (the method's own by-profile limiter) stops granting permits and answers 429 to everything. A limiter without `:queue` gets an unlimited queue and therefore never rejects: its pile-up is silent, with no 429 and no log line except the WARN `enqueued`/`acquired` lines of the keyed id.
- Deadlock shape between a command's own limits and the root ones (fixed): a command declaring its own limits held `process-image/by-profile` before `root/by-profile`, while a command with no limits of its own takes the roots first and then re-enters with the same keyed limits from inside its handler (`create-file-media-object-from-url`, `upload-photo`) and waits for them holding a root permit. Two calls of one profile then held the same two limiters inverted and parked for good: 69 threads at 0 CPU, an audit limiter with 0 acquisitions in 24h and 100% 429s, on an instance that never served a 204.
- `climit/wrap` now appends the root limiters after the command's own, so `root/global`/`root/by-profile` end up outermost and every path meets them first; the nested `climit/invoke!` chains carry only the command's own limits and run inside them, which puts both paths in the same order. `impl-invoke` also keeps, per thread, the set of limiters that thread is inside of and runs the handler inline when a call asks for one it already holds, instead of queueing behind itself. That check relies on `pbh/invoke!` running the handler on the calling thread; the mark is removed in a `finally`. Commands declare the by-profile limit before the global one, which makes it the outermost of the pair: keep that declaration order, the reverse one would put two paths back in inverted order. The two nested call sites (`create-file-media-object-from-url`, `upload-photo`) still share `process-image` with the enclosing call instead of getting a limiter of their own, so the comment there about keeping the download out of that limit no longer holds.
- The shipped `backend/resources/climit.edn` sets a timeout on every limiter, 60s for operations that can be slow and 30s otherwise, with a test that fails when an entry ships without one, and `init-key` logs a WARN for every entry without a timeout. The production file lives outside the repository: mirror it there, and remember that a parked thread only goes away with a restart or a deploy.
## DB helpers

3
.serena/memories/backend/testing.md
View File

@ -8,4 +8,5 @@ JVM `clojure.test` (kaocha runner) under `backend/test/backend_tests/`.
- Coverage: if code is added or modified in `src/`, corresponding tests in `test/backend_tests/` must be added or updated.
- Isolated run: `clojure -M:dev:test --focus backend-tests.my-ns-test` for a specific test namespace, or `clojure -M:dev:test --focus backend-tests.my-ns-test/my-test-var` for a specific test var.
- Regression run: `clojure -M:dev:test` to ensure no regressions in related functional areas.
- If you need to filter output, tee to a temp file first: `clojure -M:dev:test 2>&1 | tee /tmp/penpot-test-output.txt`.
- If you need to filter output, tee to a temp file first: `clojure -M:dev:test 2>&1 | tee /tmp/penpot-test-output.txt`.
- `backend-tests.helpers` has `metrics-stub` (a metrics instance with no collectors, where every `run!` is a no-op) and `init-climit` (a concurrency limiter built from an inline config map), for tests that need a limiter without starting the system. Log lines are capturable with `add-watch` on `app.common.logging/log-record` (it holds the last record; the watcher sees every one).

51
backend/resources/climit.edn
View File

@ -1,54 +1,65 @@
;; Example climit.edn file
;; Required: permits
;; Optional: queue, ommited means Integer/MAX_VALUE
;; Optional: timeout, ommited means no timeout
;; Note: queue and timeout are excluding
{:update-file/global {:permits 20}
;; Optional: queue, omitted means Integer/MAX_VALUE (unbounded)
;; Optional: timeout in milliseconds, omitted means the thread waits for a
;; permit forever, and a saturated limiter parks it for good. Values below:
;; 60s for operations that can be slow (file updates, images, fonts, thumbnails,
;; snapshots, imports) and 30s for the rest.
;; Note: a queue bounds how many waiters pile up, a timeout bounds how long one
;; waiter waits. Every entry sets a timeout; the by-profile ones also set a
;; queue, which is where a pile-up hurts.
{:update-file/global
{:permits 20 :timeout 60000}
:update-file/by-profile
{:permits 1 :queue 5}
{:permits 1 :queue 5 :timeout 60000}
:process-font/global {:permits 4}
:process-font/by-profile {:permits 1}
:process-font/global
{:permits 4 :timeout 60000}
:process-font/by-profile
{:permits 1 :timeout 60000}
:process-image/global {:permits 8}
:process-image/by-profile {:permits 1}
:process-image/global
{:permits 8 :timeout 60000}
:process-image/by-profile
{:permits 1 :timeout 60000}
:auth/global {:permits 8}
:auth/global
{:permits 8 :timeout 30000}
:root/global
{:permits 40}
{:permits 40 :timeout 30000}
:root/by-profile
{:permits 10 :queue 30 :timeout 30000}
:file-thumbnail-ops/global
{:permits 20}
{:permits 20 :timeout 60000}
:file-thumbnail-ops/by-profile
{:permits 2}
{:permits 2 :timeout 60000}
:submit-audit-events/by-profile
{:permits 1 :queue 3}
{:permits 1 :queue 3 :timeout 30000}
:restore-file-snapshot/global
{:permits 3}
{:permits 3 :timeout 60000}
:restore-file-snapshot/by-profile
{:permits 1 :queue 2 :timeout 60000}
:create-file-snapshot/global
{:permits 3}
{:permits 3 :timeout 60000}
:create-file-snapshot/by-profile
{:permits 1 :queue 2 :timeout 60000}
:send-user-feedback/global
{:permits 4}
{:permits 4 :timeout 30000}
:send-user-feedback/by-profile
{:permits 1 :queue 3}
{:permits 1 :queue 3 :timeout 30000}
:import-binfile/global
{:permits 4}
{:permits 4 :timeout 60000}
:import-binfile/by-profile
{:permits 1 :queue 2}}
{:permits 1 :queue 2 :timeout 60000}}

89
backend/src/app/rpc/climit.clj
View File

@ -71,6 +71,9 @@
(when-let [params (some->> config slurp edn/read-string check-config)]
(l/inf :hint "initializing concurrency limit" :config (str config))
(let [params (reduce-kv (fn [result k v]
(when-not (:timeout v)
(l/wrn :hint "limiter without timeout, a saturated one parks its threads forever"
:id k))
(assoc result k (assoc v ::id k)))
params
params)
@ -158,6 +161,16 @@
(def ^:private idseq (AtomicLong. 0))
(def ^:private ^ThreadLocal held-limiters
;; Limiters a thread is inside of, identified by the limiter instance, which
;; already encodes the limit id and its key: the cache is per id+key pair.
;; `pbh/invoke!` runs the handler on the calling thread, so a handler that
;; enters climit again for a limit it holds finds it here. Without this a
;; thread parks forever behind a permit it holds itself.
(ThreadLocal/withInitial
(reify java.util.function.Supplier
(get [_] (java.util.HashSet.)))))
(defn- impl-invoke
[metrics cache config handler]
(let [limit-id (::id config)
@ -170,38 +183,50 @@
mlabels (into-array String [(id->str limit-id)])
limit-id (id->str limit-id limit-key)
limiter (cache/get cache limit-id (partial create-limiter config))
tpoint (ct/tpoint)
req-id (.incrementAndGet ^AtomicLong idseq)]
(try
(let [stats (pbh/get-stats limiter)]
(measure metrics mlabels stats nil)
(log "enqueued" req-id stats limit-id limit-label limit-params nil))
held ^java.util.Set (.get ^ThreadLocal held-limiters)]
(pbh/invoke! limiter (fn []
(let [elapsed (tpoint)
stats (pbh/get-stats limiter)]
(measure metrics mlabels stats elapsed)
(log "acquired" req-id stats limit-id limit-label limit-params elapsed)
(handler))))
(if (.contains held limiter)
;; The calling thread already holds this limiter for this key: run the
;; handler inline instead of queueing behind itself.
(do
(l/trc :hint "reentrant invocation" :id limit-id :label limit-label)
(handler))
(catch ExceptionInfo cause
(let [{:keys [type code]} (ex-data cause)]
(if (= :bulkhead-error type)
(let [tpoint (ct/tpoint)
req-id (.incrementAndGet ^AtomicLong idseq)]
(.add held limiter)
(try
(let [stats (pbh/get-stats limiter)]
(measure metrics mlabels stats nil)
(log "enqueued" req-id stats limit-id limit-label limit-params nil))
(pbh/invoke! limiter (fn []
(let [elapsed (tpoint)
stats (pbh/get-stats limiter)]
(measure metrics mlabels stats elapsed)
(log "acquired" req-id stats limit-id limit-label limit-params elapsed)
(handler))))
(catch ExceptionInfo cause
(let [{:keys [type code]} (ex-data cause)]
(if (= :bulkhead-error type)
(let [elapsed (tpoint)
stats (pbh/get-stats limiter)]
(log "rejected" req-id stats limit-id limit-label limit-params elapsed)
(ex/raise :type :concurrency-limit
:code code
:hint "concurrency limit reached"
:cause cause))
(throw cause))))
(finally
(.remove held limiter)
(let [elapsed (tpoint)
stats (pbh/get-stats limiter)]
(log "rejected" req-id stats limit-id limit-label limit-params elapsed)
(ex/raise :type :concurrency-limit
:code code
:hint "concurrency limit reached"
:cause cause))
(throw cause))))
stats (pbh/get-stats limiter)]
(finally
(let [elapsed (tpoint)
stats (pbh/get-stats limiter)]
(measure metrics mlabels stats nil)
(log "finished" req-id stats limit-id limit-label limit-params elapsed))))))
(measure metrics mlabels stats nil)
(log "finished" req-id stats limit-id limit-label limit-params elapsed))))))))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; MIDDLEWARE
@ -263,7 +288,13 @@
(l/wrn :hint "no config found for specified queue" :id (id->str limit-id))
handler)))
handler
(concat global-limits (get-limits mdata)))
;; The root limiters go last so they end up outermost: every path
;; meets them before the command's own limits. The nested
;; `climit/invoke!` chains carry only the command's own limits and
;; run inside those of the enclosing call, so this puts both paths
;; in the same order. With the opposite order two calls of the same
;; profile hold the same limiters inverted and park for good.
(concat (get-limits mdata) global-limits))
handler))

264
backend/test/backend_tests/climit_test.clj Normal file
View File

@ -0,0 +1,264 @@
;; This Source Code Form is subject to the terms of the Mozilla Public
;; License, v. 2.0. If a copy of the MPL was not distributed with this
;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
;;
;; Copyright (c) KALEIDOS INC Sucursal en España SL
(ns backend-tests.climit-test
(:require
[app.common.logging :as l]
[app.common.uuid :as uuid]
[app.metrics :as mtx]
[app.rpc :as-alias rpc]
[app.rpc.climit :as climit]
[app.util.services :as-alias sv]
[backend-tests.helpers :as th]
[clojure.edn :as edn]
[clojure.java.io :as io]
[clojure.test :as t]))
(def ^:private !calls (atom []))
(defn- record-metrics!
"Run `f` recording every metrics call it makes."
[f]
(reset! !calls [])
(with-redefs [mtx/run!
(fn [_ & {:keys [id labels]}]
(swap! !calls conj {:id id :labels (vec labels)}))]
(f)))
(defn- acquired-in-order
"Limiter names acquired so far, in acquisition order."
[]
(->> @!calls
(filter #(= :rpc-climit-timing (:id %)))
(mapcat :labels)
(vec)))
(defn- saw-call?
"Wait until a metrics call mentions `label`."
[label timeout]
(let [deadline (+ (System/currentTimeMillis) timeout)]
(loop []
(cond
(some (fn [call] (some #{label} (:labels call))) @!calls)
true
(> (System/currentTimeMillis) deadline)
false
:else
(do (Thread/sleep 10) (recur))))))
(defn- failed-type
"Run `f` and return its value, or the type of the error it raised."
[f]
(try
(f)
(catch Throwable cause
(:type (ex-data cause)))))
(def ^:private no-timeout-hint
"limiter without timeout, a saturated one parks its threads forever")
(defn- warned-ids
"Ids of the captured log records that warn about a missing timeout."
[records]
(->> records
(keep (fn [record]
(when (= no-timeout-hint (get (::l/props record) :hint))
(get (::l/props record) :id))))
(set)))
(defn- warned?
"Wait until a captured log record warns about a missing timeout."
[records timeout]
(let [deadline (+ (System/currentTimeMillis) timeout)]
(loop []
(cond
(seq (warned-ids @records)) true
(> (System/currentTimeMillis) deadline) false
:else (do (Thread/sleep 10) (recur))))))
;; One permit and a bounded wait per limiter: a call either gets its permit, or
;; fails with a concurrency limit error, never parks a thread forever.
(def ^:private test-config
{:root/global {:permits 1 :timeout 3000}
:root/by-profile {:permits 1 :timeout 3000}
:process-image/global {:permits 1 :timeout 3000}
:process-image/by-profile {:permits 1 :timeout 3000}})
;; The limits `upload-file-media-object` and `assemble-file-media-object`
;; declare, per profile.
(def ^:private image-limits
[[:process-image/by-profile ::rpc/profile-id]
[:process-image/global]])
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; ACQUISITION ORDER
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
(t/deftest acquires-root-limits-before-the-command-own-limits
;; A command declaring its own limits must meet the root limiters first.
;; The nested `climit/invoke!` calls take only the command's own limits and
;; run inside the root ones, so any other order lets two calls of the same
;; profile hold the same limiters in opposite order and park for good.
(record-metrics!
(fn []
(let [climit (th/init-climit test-config)
mdata {::sv/name "test-command" ::climit/id image-limits}
wrapped (climit/wrap {::rpc/climit climit} (fn [_ _] :ok) mdata)
result (wrapped {::rpc/climit climit} {::rpc/profile-id (uuid/next)})]
(t/is (= :ok result))
(t/is (= ["root/by-profile"
"root/global"
"process-image/by-profile"
"process-image/global"]
(acquired-in-order)))))))
(t/deftest nested-limits-do-not-deadlock-with-the-command-own-limits
;; Two paths over one profile: a command declaring its own limits (like
;; `upload-file-media-object`) and a command that takes the root limiters and
;; then asks for the same limits from inside its handler (like
;; `create-file-media-object-from-url`). Each path holds what the other one
;; waits for the moment the order differs, and nothing releases.
(record-metrics!
(fn []
(let [climit (th/init-climit test-config)
pid (uuid/next)
gate (promise)
entered (promise)
handler-b (fn [cfg params]
(deliver entered true)
@gate
(-> cfg
(assoc ::climit/id [[:process-image/by-profile pid]
[:process-image/global]])
(assoc ::climit/label "path-b")
(climit/invoke! (fn [_ _] :nested-ok) params)))
wrapped-a (climit/wrap {::rpc/climit climit}
(fn [_ _] :ok)
{::sv/name "path-a" ::climit/id image-limits})
wrapped-b (climit/wrap {::rpc/climit climit}
handler-b
{::sv/name "path-b"})]
;; Path B takes the root limiters and waits inside its handler.
(let [fb (future (failed-type #(wrapped-b {::rpc/climit climit}
{::rpc/profile-id pid})))]
(t/is (true? (deref entered 5000 false)))
;; Path A, started now, takes its own limiters and then waits for the
;; root ones, which B holds. The recorded metrics are how the test
;; knows A got that far instead of guessing with a sleep.
(reset! !calls [])
(let [fa (future (failed-type #(wrapped-a {::rpc/climit climit}
{::rpc/profile-id pid})))]
(t/is (true? (saw-call? "root/by-profile" 5000)))
;; Let B ask for the limits A holds: if the order differs, each path
;; now waits for a permit the other one holds.
(deliver gate true)
(t/is (= :ok (deref fa 15000 :timeout)))
(t/is (= :nested-ok (deref fb 15000 :timeout)))))))))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; REENTRANCY
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
(t/deftest reentrant-invoke-does-not-consume-a-second-permit
;; A handler entering climit for a limit it already holds must not queue
;; behind itself: it runs inline and takes no second permit. Another thread
;; asking for the same limiter still waits, so the limit stays in place.
(let [climit (th/init-climit {:process-image/by-profile
{:permits 1 :timeout 2000}})
pid (uuid/next)
cfg {::rpc/climit climit
::climit/id [[:process-image/by-profile pid]]}
inner (promise)
other (promise)
handler (fn [cfg params]
(deliver inner
(failed-type #(climit/invoke! cfg (fn [_ _] :inner-ok) params)))
(deliver other (future (climit/invoke! cfg (fn [_ _] :other-ok) params)))
(t/is (= :waiting (deref @other 300 :waiting)))
:outer-ok)
outcome (failed-type #(climit/invoke! cfg handler {}))]
;; With no reentrancy handling the inner call waits for the permit this very
;; thread holds and ends failing by timeout.
(t/is (= :inner-ok (deref inner 5000 :timeout)))
(t/is (= :outer-ok outcome))
;; once the outer call releases, the other thread gets its permit
(t/is (= :other-ok (deref @other 5000 :timeout)))))
(t/deftest reentrant-tracking-is-cleaned-up-when-the-handler-fails
;; If the mark is not removed, the thread keeps believing it holds the
;; limiter and every later call of that key would skip the limit.
(let [climit (th/init-climit {:process-image/by-profile
{:permits 1 :timeout 1000}})
pid (uuid/next)
cfg {::rpc/climit climit
::climit/id [[:process-image/by-profile pid]]}
holding (promise)
gate (promise)]
;; The failing call runs on this thread, so this thread is the one that
;; must forget it held the limiter.
(t/is (thrown? clojure.lang.ExceptionInfo
(climit/invoke! cfg (fn [_ _] (throw (ex-info "boom" {}))) {})))
;; another thread takes the only permit
(let [fb (future (climit/invoke! cfg (fn [_ _]
(deliver holding true)
@gate
:held)
{}))]
(t/is (true? (deref holding 5000 false)))
;; this thread must wait for it, not skip the limiter
(t/is (= :concurrency-limit
(failed-type #(climit/invoke! cfg (fn [_ _] :inner-ok) {}))))
(deliver gate true)
(t/is (= :held (deref fb 5000 :timeout))))))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; SHIPPED CONFIG
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
(t/deftest example-config-bounds-every-wait
;; A limiter without `:timeout` parks the calling thread forever, which is
;; how a limit nobody releases turns into lost worker threads and days of
;; silent 429s. Every entry of the shipped config must carry one.
(let [path (io/resource "climit.edn")]
(t/is (some? path))
(let [data (edn/read-string (slurp path))]
(t/is (seq data))
(doseq [[id config] data]
(t/is (int? (:timeout config))
(str id " must declare a timeout"))))))
(t/deftest warns-when-a-limiter-has-no-timeout
;; The config that production loads lives outside the repository and is
;; maintained by hand, so a limiter without a timeout has to say so at
;; startup instead of parking threads in silence months later.
(let [!records (atom [])]
(add-watch l/log-record ::capture
(fn [_ _ _ record] (swap! !records conj record)))
(try
(th/init-climit {:process-image/by-profile {:permits 1 :timeout 1000}
:process-image/global {:permits 8}})
(t/is (true? (warned? !records 2000)))
(t/is (= #{:process-image/global} (warned-ids @!records)))
(finally
(remove-watch l/log-record ::capture)))))

21
backend/test/backend_tests/helpers.clj
View File

@ -23,9 +23,11 @@
[app.main :as main]
[app.media]
[app.media :as-alias mtx]
[app.metrics :as-alias metrics]
[app.migrations]
[app.msgbus :as-alias mbus]
[app.rpc :as-alias rpc]
[app.rpc.climit :as-alias climit]
[app.rpc.commands.auth :as cmd.auth]
[app.rpc.commands.files :as files]
[app.rpc.commands.files-create :as files.create]
@ -425,6 +427,25 @@
(assoc :app.worker.runner/id 0))]
(#'app.worker.runner/run-task cfg task)))))))
(defn metrics-stub
"A metrics instance with no collectors, where every `run!` is a no-op."
[]
(reify metrics/IMetrics
(get-handler [_] nil)
(get-collector [_ _] nil)
(get-registry [_] nil)))
(defn init-climit
"Build a concurrency limiter component from an inline config map."
[config]
(let [file (doto (java.io.File/createTempFile "climit" ".edn")
(.deleteOnExit))]
(spit file (pr-str config))
(ig/init-key ::rpc/climit
{::climit/enabled true
::climit/config (.getAbsolutePath file)
::metrics/metrics (metrics-stub)})))
;; --- UTILS
(defn print-error!

19
backend/test/backend_tests/rpc_audit_test.clj
View File

@ -12,13 +12,11 @@
[app.config :as cf]
[app.db :as db]
[app.loggers.audit :as audit]
[app.metrics :as mtx]
[app.rpc :as-alias rpc]
[app.rpc.climit :as climit]
[app.util.services :as sv]
[backend-tests.helpers :as th]
[clojure.test :as t]
[integrant.core :as ig]
[yetti.request]))
(t/use-fixtures :once th/state-init)
@ -45,21 +43,8 @@
;; `key-fn` of the vector form. That silently turned the limit into a single
;; shared one, so one stuck call ended up rejecting the audit batches of
;; every profile on the instance. Two profiles must not share a limiter.
(let [config (doto (java.io.File/createTempFile "climit" ".edn")
(.deleteOnExit))
_ (spit config "{:submit-audit-events/by-profile {:permits 1 :queue 1}}")
;; The limiter only uses the metrics instance to measure, and with a
;; stub collector every `run!` is a no-op.
metrics (reify mtx/IMetrics
(get-handler [_] nil)
(get-collector [_ _] nil)
(get-registry [_] nil))
climit (ig/init-key ::rpc/climit
{::climit/enabled true
::climit/config (.getAbsolutePath config)
::mtx/metrics metrics})
(let [climit (th/init-climit {:submit-audit-events/by-profile
{:permits 1 :queue 1 :timeout 30000}})
;; The metadata declared by the command itself, that is what the test
;; has to cover.

51
backend/test/backend_tests/rpc_climit_test.clj Normal file
View File

@ -0,0 +1,51 @@
;; This Source Code Form is subject to the terms of the Mozilla Public
;; License, v. 2.0. If a copy of the MPL was not distributed with this
;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
;;
;; Copyright (c) KALEIDOS INC Sucursal en España SL
(ns backend-tests.rpc-climit-test
(:require
[app.rpc.climit :as-alias climit]
[app.util.services :as-alias sv]
[backend-tests.helpers :as th]
[clojure.test :as t]))
(t/use-fixtures :once th/state-init)
(defn- method-limit-declarations
"The `::climit/id` declarations of every RPC command, as the running system
wraps them."
[]
(->> (:app.rpc/methods th/*system*)
(vals)
(map first)
(keep (fn [mdata]
(when-let [limits (::climit/id mdata)]
[mdata limits])))))
(t/deftest commands-declare-the-by-profile-limit-before-the-global-one
;; `climit/get-limits` reverses a declared vector of pairs, so the first one
;; declared is acquired first, and the nested `climit/invoke!` chains read the
;; same declaration. A pair declared global first puts a command back in the
;; inverted order that parked two paths of one profile for good.
(let [declarations (method-limit-declarations)]
;; the system wraps every command, so a short list means the scan went stale
(t/is (<= 10 (count declarations)))
(doseq [[mdata limits] declarations
:when (and (vector? limits) (vector? (first limits)))]
(let [ids (mapv first limits)
index (into {} (map-indexed (fn [idx id] [id idx]) ids))
where (str (::sv/name mdata))]
(t/is (= (count ids) (count (distinct ids)))
(str where " declares the same limit twice"))
(doseq [[id idx] index
:when (= "by-profile" (name id))]
(when-let [gidx (get index (keyword (namespace id) "global"))]
(t/is (< idx gidx)
(str where " must declare " id " before "
(keyword (namespace id) "global")))))))))