diff --git a/.serena/memories/backend/rpc-db-worker-subtleties.md b/.serena/memories/backend/rpc-db-worker-subtleties.md index d72868e00a..26e07651f4 100644 --- a/.serena/memories/backend/rpc-db-worker-subtleties.md +++ b/.serena/memories/backend/rpc-db-worker-subtleties.md @@ -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 diff --git a/.serena/memories/backend/testing.md b/.serena/memories/backend/testing.md index 66d6f8a246..8dc747ab14 100644 --- a/.serena/memories/backend/testing.md +++ b/.serena/memories/backend/testing.md @@ -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`. \ No newline at end of file +- 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). \ No newline at end of file diff --git a/backend/resources/climit.edn b/backend/resources/climit.edn index ded9b5c9b8..8c0a743319 100644 --- a/backend/resources/climit.edn +++ b/backend/resources/climit.edn @@ -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}} diff --git a/backend/src/app/rpc/climit.clj b/backend/src/app/rpc/climit.clj index 60d14af09b..b8b49b0cda 100644 --- a/backend/src/app/rpc/climit.clj +++ b/backend/src/app/rpc/climit.clj @@ -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)) diff --git a/backend/test/backend_tests/climit_test.clj b/backend/test/backend_tests/climit_test.clj new file mode 100644 index 0000000000..acf198fb84 --- /dev/null +++ b/backend/test/backend_tests/climit_test.clj @@ -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))))) diff --git a/backend/test/backend_tests/helpers.clj b/backend/test/backend_tests/helpers.clj index a9dd895176..ccf11466ed 100644 --- a/backend/test/backend_tests/helpers.clj +++ b/backend/test/backend_tests/helpers.clj @@ -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! diff --git a/backend/test/backend_tests/rpc_audit_test.clj b/backend/test/backend_tests/rpc_audit_test.clj index 2dea7e2149..fa82d8fb40 100644 --- a/backend/test/backend_tests/rpc_audit_test.clj +++ b/backend/test/backend_tests/rpc_audit_test.clj @@ -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. diff --git a/backend/test/backend_tests/rpc_climit_test.clj b/backend/test/backend_tests/rpc_climit_test.clj new file mode 100644 index 0000000000..b5b8e853d8 --- /dev/null +++ b/backend/test/backend_tests/rpc_climit_test.clj @@ -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")))))))))