From 3cc91e8ba218f3ee068429a38a83a10f8c3f7c32 Mon Sep 17 00:00:00 2001 From: Andrey Antukh Date: Wed, 28 Oct 2026 18:44:36 +0100 Subject: [PATCH] :recycle: Take the command limits before the root ones in the RPC limiter (#12686) * :recycle: Take the command limits before the root ones in the RPC limiter The RPC concurrency limiter took the root limiters (the ones every request passes through) before the ones a command declares, so a request waiting for the limits of its own command held the root permits of its profile while it waited: a profile with a pile-up of heavy requests saw its other requests queue behind them and be rejected with 429. That order was forced by two commands that entered the limiter from inside their handler: a command that declares no limits of its own can only take them deeper than the root ones, so two paths of one profile could take the same limiters in opposite order and park for good. The commands declare their limits in their metadata now and their handlers call the work directly, so no command enters the limiter from inside its handler and every path takes the limiters in the same order. The root limiters are appended to the sequence first, so the limits a command declares end up outermost and are taken first: a request waiting for them does not hold the root permits of its profile. The photo of a team keeps its limits, because it declares them now that the helper does not. A call from inside a handler can only reenter the limits that thread already holds, and it is refused when it asks for a new one, because that would take it after the root ones and invert the order against the other paths. The key of a limit is resolved the same way in every path, so both paths point at the same limiter and the reentrancy check sees it. Every entry of the shipped climit.edn declares a queue, so the number of requests waiting for the same limit is bounded instead of piling up until the timeout. The production file lives outside the repository and has to be mirrored to pick them up. The tests cover the acquisition order of a command with and without limits of its own, the order of the limits as a whole (a ring of any size fails, not only a pair in opposite order), the nested call that asks for a limit the thread does not hold, and the queues and timeouts of the shipped configuration. The import of a binfile declares its request as multipart in its test, the way the commands that take an upload map do. AI-assisted-by: minimax-m3.1-flash AI-assisted-by: deepseek-v4.1-flash * :wrench: Keep the commands' limits and the shipped config in sync `profile-plugin-ops` guarded only an UPDATE on the caller's own profile row, which the row lock already serializes, so its global cap could only reject a burst of different profiles toggling plugins. Both commands declare the by-profile limit alone now, and the example config drops the entry. A declared id without an entry does not limit anything: `wrap` logs a warning and runs the handler with no limiter at all, which is how a command ends up unbounded in production. A test fails now when a command declares an id the config does not list, and the header of the example says the production file mirrors the shape and lists every declared id, not the values. AI-assisted-by: deepseek-v4.1-flash * :books: Record that background work is not waited for with a sleep A long command moved to the background reports its own completion, so polling it with a sleep inside a foreground command only blocks the session and the notification arrives anyway. The rule says to keep working on something else, or to end the response and wait for it. AI-assisted-by: minimax-m3.1-flash --- .serena/memories/backend/subtleties.md | 7 +- .serena/memories/critical-info.md | 1 + backend/resources/climit.edn | 38 ++- backend/src/app/rpc/climit.clj | 174 ++++------ backend/src/app/rpc/commands/media.clj | 13 +- backend/src/app/rpc/commands/plugins.clj | 6 +- backend/src/app/rpc/commands/profile.clj | 15 +- backend/src/app/rpc/commands/teams.clj | 3 + backend/test/backend_tests/climit_test.clj | 307 +++++++----------- backend/test/backend_tests/helpers.clj | 10 +- .../test/backend_tests/rpc_binfile_test.clj | 16 +- .../test/backend_tests/rpc_climit_test.clj | 134 +++++++- 12 files changed, 379 insertions(+), 345 deletions(-) diff --git a/.serena/memories/backend/subtleties.md b/.serena/memories/backend/subtleties.md index e778f31f8e..1b4fa4ddec 100644 --- a/.serena/memories/backend/subtleties.md +++ b/.serena/memories/backend/subtleties.md @@ -17,9 +17,10 @@ - The waiting counters are a field of the `ConcurrencyLimiter` component (a `ConcurrentHashMap` of name to `AtomicLong`), so they start at zero, die with the component, and the sampler that closes over that map republishes only its own names every 15 s; the sampler stops with the component in `halt-key!`. `impl-invoke` pairs one decrement with one increment through a per-call flag: the flag turns a second call into a no-op and the `finally` guarantees that there is one. The increment sits inside the `try` and the decrement lives only in the acquired callback and the `finally`, so a handler that throws, a rejection and a throwable from the publishing path all leave the count where it was and the thread without the limiter marked as held (a leftover mark would skip that limiter for that key for ever). Concurrent `inc`/`dec` publications of two threads can land out of order and the value can read low for a moment; the republish corrects it. - The `app.rpc.climit` log lines carry `queue`, `queue-max`, `permits`, `elapsed`, `code` and `reason`, and their `id` carries the key (`name/`). That is the only place where the profile behind a queue is visible: a pile-up is chased there, on demand, not in the metrics. The `queue` of the line is the calls inside the limiter minus its permits, so it reads 0 for a lone caller that times out: that line goes out at TRACE instead of WARN, and the rejection count has to come from the metric. - 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. +- 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. Both commands declare their limits now, so the shape cannot happen any more: the root limits are taken last and every path takes the limiters in the same order. +- `climit/wrap` takes the command's own limits before the root ones: the root limiters are appended to the sequence first and the last element of the sequence is the outermost one, so `root/global`/`root/by-profile` end up innermost. A request waiting for its own limits no longer holds the root permits of its profile, which is what kept a pile-up of heavy requests from blocking the other commands of that profile. Every acquisition goes through that wrapper, so every path takes the limiters in the same order: there is no way to enter the limiter from inside a handler, and a command that calls another command internally would take the limiters twice (the second call waits for a permit the first one holds and answers 429 when the timeout runs out). The key of a limit is a function of the params (a keyword is looked up in them, a function is applied to them, any other value is the key itself). Commands declare the by-profile limit before the global one, which makes it the outermost of the pair, and a test fails when the relation between all the limits the commands declare cannot be sorted: two commands that take the same limits in opposite order, or a ring of three, leave the paths of one profile holding them in opposite order. +- The two commands that used to enter climit from inside their handler declare their limits now, so the whole command runs inside them: the download of an external image takes a `process-image` permit, and that limit has one permit per profile, so a slow download holds the image processing of that profile for as long as it takes. It takes the global permit of that limit too (8 permits, queue 20), so eight slow downloads of eight different profiles leave the image processing of every profile waiting. The copy of the body has no timeout (the 30 s of the request covers the connection and the headers), so a server that stops sending keeps the permits until the operating system gives up. The storage write of the profile photo runs inside its limit too. +- The shipped `backend/resources/climit.edn` sets a timeout and a queue 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 queue matters because the root limits no longer bound the waiters: a command takes its own limits first, so a pile-up waits in the queue of that limit. The file and the declarations have to list the same ids: a `::climit/id` without an entry makes `wrap` log `no config found for specified queue` and run the handler with no limiter at all, and a test fails when a command declares an id the file does not list. Only key a limit by profile when the work contends on something shared (`update-file`, images, fonts, thumbnails, the profile row): a global cap on a cheap command adds a rejection path nobody needs, which is why `profile-plugin-ops` and `submit-audit-events` declare only `by-profile`. 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/critical-info.md b/.serena/memories/critical-info.md index d66a46eb84..00eb2cab6f 100644 --- a/.serena/memories/critical-info.md +++ b/.serena/memories/critical-info.md @@ -25,6 +25,7 @@ You are working on the GitHub project `penpot/penpot`, a monorepo. - Align `let` binding values: when a `let` form has multiple bindings spanning several lines, align the value forms to the same column with spaces. - If you introduce delimiter errors (mismatched parens/brackets) in Clojure/CLJS files, fix them with `scripts/paren-repair` BEFORE running lint/format checks. See `mem:scripts/paren-repair` for usage. - Never run anything that destroys data without explicit permission, including `drop-devenv`, `docker compose down -v`, `docker volume rm ...`. The user's real work lives in the volumes of the shared infra. +- **Never wait for background work with a sleep.** A long command moved to the background reports its own completion: do not poll it with `sleep 150` (or any other wait) inside a foreground command, and do not re-read its output file to check whether it finished. Keep working on something else, or end the response and wait for the notification. # Project modules diff --git a/backend/resources/climit.edn b/backend/resources/climit.edn index 9b585edabc..1a5c8f1ecc 100644 --- a/backend/resources/climit.edn +++ b/backend/resources/climit.edn @@ -6,66 +6,68 @@ ;; 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. +;; waiter waits. Every entry sets a timeout and a queue: a command takes the +;; limits it declares before the root ones, so the queue of a limit is where +;; its own pile-up waits. The root limits bound what runs at the same time, +;; not what waits for it. +;; The production file lives outside this repository: mirror the shape there, +;; not these numbers. It has to list every id a command declares: an id without +;; an entry runs with no limiter at all. {:update-file/global - {:permits 20 :timeout 60000} + {:permits 20 :queue 50 :timeout 60000} :update-file/by-profile {:permits 1 :queue 5 :timeout 60000} :process-font/global - {:permits 4 :timeout 60000} + {:permits 4 :queue 10 :timeout 60000} :process-font/by-profile - {:permits 1 :timeout 60000} + {:permits 1 :queue 5 :timeout 60000} :process-image/global - {:permits 8 :timeout 60000} + {:permits 8 :queue 20 :timeout 60000} :process-image/by-profile - {:permits 1 :timeout 60000} + {:permits 1 :queue 5 :timeout 60000} :auth/global - {:permits 8 :timeout 30000} + {:permits 8 :queue 20 :timeout 30000} :root/global - {:permits 40 :timeout 30000} + {:permits 40 :queue 100 :timeout 30000} :root/by-profile {:permits 10 :queue 30 :timeout 30000} :file-thumbnail-ops/global - {:permits 20 :timeout 60000} + {:permits 20 :queue 50 :timeout 60000} :file-thumbnail-ops/by-profile - {:permits 2 :timeout 60000} + {:permits 2 :queue 10 :timeout 60000} :submit-audit-events/by-profile {:permits 1 :queue 3 :timeout 30000} :restore-file-snapshot/global - {:permits 3 :timeout 60000} + {:permits 3 :queue 5 :timeout 60000} :restore-file-snapshot/by-profile {:permits 1 :queue 2 :timeout 60000} :create-file-snapshot/global - {:permits 3 :timeout 60000} + {:permits 3 :queue 5 :timeout 60000} :create-file-snapshot/by-profile {:permits 1 :queue 2 :timeout 60000} :send-user-feedback/global - {:permits 4 :timeout 30000} + {:permits 4 :queue 10 :timeout 30000} :send-user-feedback/by-profile {:permits 1 :queue 3 :timeout 30000} :import-binfile/global - {:permits 4 :timeout 60000} + {:permits 4 :queue 10 :timeout 60000} :import-binfile/by-profile {:permits 1 :queue 2 :timeout 60000} - :profile-plugin-ops/global - {:permits 4 :timeout 30000} - :profile-plugin-ops/by-profile {:permits 1 :queue 2 :timeout 30000}} diff --git a/backend/src/app/rpc/climit.clj b/backend/src/app/rpc/climit.clj index 82fe9ff234..3f7330238e 100644 --- a/backend/src/app/rpc/climit.clj +++ b/backend/src/app/rpc/climit.clj @@ -5,7 +5,16 @@ ;; Copyright (c) KALEIDOS SUBSIDIARY SL (ns app.rpc.climit - "Concurrencly limiter for RPC." + "Concurrencly limiter for RPC. + + A command declares its limits in its `::climit/id` metadata and takes them + before the root ones, so a request waiting for its own limits does not hold + the root permits of its profile. + + Every acquisition goes through this wrapper, so every path takes the + limiters in the same order. A command that calls another command from + inside its handler would take them twice: the second call waits for a + permit the first one holds, and answers 429 when the timeout runs out." (:refer-clojure :exclude [run!]) (:require [app.common.exceptions :as ex] @@ -221,16 +230,6 @@ (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 counters cache config handler] (let [limit-id (::id config) @@ -249,69 +248,57 @@ :permits (or (:permits config) (:concurrency config) 1) :queue-max (or (:queue config) (:queue-size config))} limiter (cache/get cache limit-id (partial create-limiter config)) - held ^java.util.Set (.get ^ThreadLocal held-limiters)] + tpoint (ct/tpoint) + req-id (.incrementAndGet ^AtomicLong idseq) + waiting (volatile! true) + stop-waiting (fn [] + ;; Exactly one decrement per increment: the flag + ;; makes a second call a no-op and the `finally` + ;; guarantees that there is one. + (when @waiting + (vreset! waiting false) + (waiting-dec metrics counters limit-name)))] - (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)) + (try + ;; The increment lives inside the `try`, so a throwable from the + ;; publishing path unwinds through the `finally`: the count comes back + ;; down instead of staying up for good. + (waiting-inc metrics counters limit-name) - (let [tpoint (ct/tpoint) - req-id (.incrementAndGet ^AtomicLong idseq) - waiting (volatile! true) - stop-waiting (fn [] - ;; Exactly one decrement per increment: the flag - ;; makes a second call a no-op and the `finally` - ;; guarantees that there is one. - (when @waiting - (vreset! waiting false) - (waiting-dec metrics counters limit-name)))] + (let [stats (pbh/get-stats limiter)] + (log limit "enqueued" req-id stats nil nil nil)) - (.add held limiter) + (pbh/invoke! limiter (fn [] + (stop-waiting) + (let [elapsed (tpoint) + stats (pbh/get-stats limiter)] + (measure metrics mlabels elapsed) + (log limit "acquired" req-id stats elapsed nil nil) + (handler)))) - (try - ;; The increment lives inside the `try`, so a throwable from the - ;; publishing path unwinds through the `finally`: the thread drops the - ;; mark and the count comes back down instead of staying up for good. - (waiting-inc metrics counters limit-name) - - (let [stats (pbh/get-stats limiter)] - (log limit "enqueued" req-id stats nil nil nil)) - - (pbh/invoke! limiter (fn [] - (stop-waiting) - (let [elapsed (tpoint) - stats (pbh/get-stats limiter)] - (measure metrics mlabels elapsed) - (log limit "acquired" req-id stats elapsed nil nil) - (handler)))) - - (catch ExceptionInfo cause - (let [{:keys [type code] :as data} (ex-data cause)] - (if (= :bulkhead-error type) - (let [elapsed (tpoint) - stats (pbh/get-stats limiter) - reason (if (some? (:timeout data)) "timeout" "queue-full")] - (mtx/run! metrics - :id :rpc-climit-rejected - :inc 1 - :labels (into-array String [limit-name reason])) - (log limit "rejected" req-id stats elapsed code reason) - (ex/raise :type :concurrency-limit - :code code - :hint "concurrency limit reached" - :cause cause)) - (throw cause)))) - - (finally - (.remove held limiter) - (stop-waiting) + (catch ExceptionInfo cause + (let [{:keys [type code] :as data} (ex-data cause)] + (if (= :bulkhead-error type) (let [elapsed (tpoint) - stats (pbh/get-stats limiter)] + stats (pbh/get-stats limiter) + reason (if (some? (:timeout data)) "timeout" "queue-full")] + (mtx/run! metrics + :id :rpc-climit-rejected + :inc 1 + :labels (into-array String [limit-name reason])) + (log limit "rejected" req-id stats elapsed code reason) + (ex/raise :type :concurrency-limit + :code code + :hint "concurrency limit reached" + :cause cause)) + (throw cause)))) - (log limit "finished" req-id stats elapsed nil nil)))))))) + (finally + (stop-waiting) + (let [elapsed (tpoint) + stats (pbh/get-stats limiter)] + + (log limit "finished" req-id stats elapsed nil nil)))))) ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; ;; MIDDLEWARE @@ -340,6 +327,16 @@ :else (throw (IllegalArgumentException. "unable to normalize limit"))))) +(defn- resolve-limit-key + "The key of a limit, resolved the same way in every path: a keyword is + looked up in the params, a function is applied to them, any other value is + the key itself." + [key params] + (cond + (keyword? key) (get params key) + (fn? key) (key params) + :else key)) + (defn wrap [cfg handler {label ::sv/name :as mdata}] (if-let [climit (::rpc/climit cfg)] @@ -363,7 +360,7 @@ (fn [cfg params] (let [config (-> config - (assoc ::key (key-fn params)) + (assoc ::key (resolve-limit-key key-fn params)) (assoc ::label label) ;; NOTE: only used for debugging output (assoc ::params params))] @@ -373,40 +370,11 @@ (l/wrn :hint "no config found for specified queue" :id (id->str limit-id)) handler))) handler - ;; 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)) + ;; The root limiters go first so they end up innermost: a command + ;; takes the limits it declares before the root ones. Every path + ;; meets the limiters in the same order, because every acquisition + ;; goes through this wrapper. + (concat global-limits (get-limits mdata))) handler)) -;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; -;; PUBLIC API -;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; - -(defn- build-exec-chain - [{:keys [::label ::rpc/climit] :as cfg} f] - (reduce (fn [handler [limit-id limit-key]] - (if-let [config (get-config climit limit-id)] - (let [config (-> config - (assoc ::key limit-key) - (assoc ::label label))] - (fn [cfg params] - (let [config (assoc config ::params params)] - (invoke climit config (partial handler cfg params))))) - (do - (l/wrn :hint "config not found" :label label :id limit-id) - f))) - f - (get-limits cfg))) - -(defn invoke! - "Run a function in context of climit." - [{:keys [::rpc/climit] :as cfg} f params] - (let [f (if climit - (build-exec-chain cfg f) - f)] - (f cfg params))) diff --git a/backend/src/app/rpc/commands/media.clj b/backend/src/app/rpc/commands/media.clj index c5a78df076..3ba28923b0 100644 --- a/backend/src/app/rpc/commands/media.clj +++ b/backend/src/app/rpc/commands/media.clj @@ -239,6 +239,8 @@ (sv/defmethod ::create-file-media-object-from-url {::doc/added "1.17" ::doc/changes [["2.19" "The optional :id param is rejected with a params-validation error; the server always generates the identifier"]] + ::climit/id [[:process-image/by-profile ::rpc/profile-id] + [:process-image/global]] ::sm/params schema:create-file-media-object-from-url} [{:keys [::db/pool] :as cfg} {:keys [::rpc/profile-id file-id] :as params}] (files/check-edition-permissions! pool profile-id file-id) @@ -262,16 +264,7 @@ (assoc :from-url? true) (assoc :content content) (assoc :name (d/nilv name "unknown")))] - - ;; NOTE: we use the climit here in a dynamic invocation because we - ;; don't want saturate the process-image limit with IO (download - ;; of external image) - - (-> cfg - (assoc ::climit/id [[:process-image/by-profile (:profile-id params)] - [:process-image/global]]) - (assoc ::climit/label "create-file-media-object-from-url") - (climit/invoke! #(db/run! %1 create-file-media-object %2) params)))) + (db/run! cfg create-file-media-object params))) ;; --- Clone File Media object (Upload and create from url) diff --git a/backend/src/app/rpc/commands/plugins.clj b/backend/src/app/rpc/commands/plugins.clj index 8f01b3b521..882ab5e597 100644 --- a/backend/src/app/rpc/commands/plugins.clj +++ b/backend/src/app/rpc/commands/plugins.clj @@ -34,8 +34,7 @@ (sv/defmethod ::add-profile-plugin {::doc/added "2.18" - ::climit/id [[:profile-plugin-ops/by-profile ::rpc/profile-id] - [:profile-plugin-ops/global]] + ::climit/id [[:profile-plugin-ops/by-profile ::rpc/profile-id]] ::sm/params schema:add-profile-plugin ::sm/result ctp/schema:registry-entry ::db/transaction true} @@ -69,8 +68,7 @@ (sv/defmethod ::remove-profile-plugin {::doc/added "2.18" - ::climit/id [[:profile-plugin-ops/by-profile ::rpc/profile-id] - [:profile-plugin-ops/global]] + ::climit/id [[:profile-plugin-ops/by-profile ::rpc/profile-id]] ::sm/params schema:remove-profile-plugin ::sm/result :nil ::db/transaction true} diff --git a/backend/src/app/rpc/commands/profile.clj b/backend/src/app/rpc/commands/profile.clj index c0cdf56971..339509b1ed 100644 --- a/backend/src/app/rpc/commands/profile.clj +++ b/backend/src/app/rpc/commands/profile.clj @@ -306,6 +306,8 @@ (sv/defmethod ::update-profile-photo {:doc/added "1.1" + ::climit/id [[:process-image/by-profile ::rpc/profile-id] + [:process-image/global]] ::sm/params schema:update-profile-photo ::rpc/multipart-only-params [:file] ::sm/result :nil} @@ -355,13 +357,12 @@ :content-type (:mtype thumb)})) (defn upload-photo - [{:keys [::sto/storage] :as cfg} {:keys [file] :as params}] - (let [params (-> cfg - (assoc ::climit/id [[:process-image/by-profile (:profile-id params)] - [:process-image/global]]) - (assoc ::climit/label "upload-photo") - (climit/invoke! generate-thumbnail file))] - (sto/put-object! storage params))) + "Generate the thumbnail of a photo and store it. + + The caller declares the `process-image` limits: this helper runs inside + them, so a command that calls it has to take them." + [{:keys [::sto/storage] :as cfg} {:keys [file]}] + (sto/put-object! storage (generate-thumbnail cfg file))) ;; --- MUTATION: Delete Photo diff --git a/backend/src/app/rpc/commands/teams.clj b/backend/src/app/rpc/commands/teams.clj index f8b85ac0eb..3b1785830a 100644 --- a/backend/src/app/rpc/commands/teams.clj +++ b/backend/src/app/rpc/commands/teams.clj @@ -26,6 +26,7 @@ [app.msgbus :as mbus] [app.nitrate :as nitrate] [app.rpc :as-alias rpc] + [app.rpc.climit :as-alias climit] [app.rpc.commands.profile :as profile] [app.rpc.doc :as-alias doc] [app.rpc.helpers :as rph] @@ -1084,6 +1085,8 @@ (sv/defmethod ::update-team-photo {::doc/added "1.17" + ::climit/id [[:process-image/by-profile ::rpc/profile-id] + [:process-image/global]] ::sm/params schema:update-team-photo ::rpc/multipart-only-params [:file]} [cfg {:keys [::rpc/profile-id file] :as params}] diff --git a/backend/test/backend_tests/climit_test.clj b/backend/test/backend_tests/climit_test.clj index b5cef6bda3..76aff9252a 100644 --- a/backend/test/backend_tests/climit_test.clj +++ b/backend/test/backend_tests/climit_test.clj @@ -38,20 +38,16 @@ (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- invoke! + "Run `f` under the limits the cfg declares, the way a command does: the key + of each limit comes from the params, and the root limiters come after the + ones the cfg declares." + [cfg f params] + (let [wrapped (climit/wrap cfg + f + {::sv/name "test-command" + ::climit/id (::climit/id cfg)})] + (wrapped cfg params))) (defn- failed-type "Run `f` and return its value, or the type of the error it raised." @@ -101,11 +97,10 @@ ;; 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. +(t/deftest acquires-the-command-own-limits-before-the-root-limits + ;; A command declaring its own limits must meet them before the root + ;; limiters. No command enters climit from inside its handler any more, so + ;; every path takes the limiters in the same order. (record-metrics (fn [] (th/with-climit [climit test-config] @@ -113,124 +108,57 @@ 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"] + (t/is (= ["process-image/by-profile" + "process-image/global" + "root/by-profile" + "root/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. +(t/deftest the-key-of-a-limit-comes-from-the-params + ;; The key of a limit is a function of the params: a keyword is looked up in + ;; them, and that is what tells one profile from another. Two calls with the + ;; same key share the limiter, and two calls with a different key do not. + (th/with-climit [climit {:test-key/by-profile {:permits 1 :timeout 2000}}] + (let [pid (uuid/next) + limits [[:test-key/by-profile ::rpc/profile-id]] + cfg {::rpc/climit climit + ::climit/id limits} + gate (promise) + entered (promise) + wrapped (climit/wrap cfg + (fn [_ params] + ;; only the call of the first profile waits + (when (= pid (::rpc/profile-id params)) + (deliver entered true) + @gate) + :ok) + {::sv/name "test-command" ::climit/id limits}) + holder (future (wrapped cfg {::rpc/profile-id pid}))] + + ;; The call of the same profile waits for the permit this thread holds. + (t/is (true? (deref entered 5000 false))) + (t/is (= :timeout (deref (future (wrapped cfg {::rpc/profile-id pid})) 300 :timeout))) + + ;; The call of another profile takes the permit of its own key. + (t/is (= :ok (deref (future (wrapped cfg {::rpc/profile-id (uuid/next)})) 5000 :timeout))) + + (deliver gate true) + (t/is (= :ok (deref holder 5000 :timeout)))))) + +(t/deftest a-command-without-limits-takes-the-root-limits + ;; A command that declares no limits of its own still takes the root ones, + ;; and in the same order as the ones of a command that declares them. (record-metrics (fn [] (th/with-climit [climit test-config] - (let [pid (uuid/next) - gate (promise) - entered (promise) + (let [mdata {::sv/name "test-command"} + 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"] + (acquired-in-order)))))))) - 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. - (th/with-climit [climit {:process-image/by-profile - {:permits 1 :timeout 2000}}] - (let [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. - (th/with-climit [climit {:process-image/by-profile - {:permits 1 :timeout 1000}}] - (let [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 @@ -239,14 +167,19 @@ (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. + ;; silent 429s. Every entry of the shipped config must carry one. A limiter + ;; without `:queue` lets its waiters pile up without bound, and a command + ;; takes its own limits before the root ones, so the queue of a limit is + ;; where its own pile-up waits. (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")))))) + (str id " must declare a timeout")) + (t/is (int? (:queue config)) + (str id " must declare a queue")))))) (t/deftest warns-when-a-limiter-has-no-timeout ;; The config that production loads lives outside the repository and is @@ -321,17 +254,17 @@ (let [name (limiter-name :test-waiting/by-profile) entered (promise) gate (promise) - holder (future (climit/invoke! (limiter-cfg climit :test-waiting/by-profile "a") - (fn [_ _] (deliver entered true) @gate :held) - {}))] + holder (future (invoke! (limiter-cfg climit :test-waiting/by-profile "a") + (fn [_ _] (deliver entered true) @gate :held) + {}))] (t/is (true? (deref entered 5000 false))) ;; the holder is running, not waiting (reset! calls []) - (let [waiter (future (climit/invoke! (limiter-cfg climit :test-waiting/by-profile "a") - (fn [_ _] :waited) - {}))] + (let [waiter (future (invoke! (limiter-cfg climit :test-waiting/by-profile "a") + (fn [_ _] :waited) + {}))] (t/is (true? (wait-until #(= 1 (waiting-last name)) 5000))) (t/is (= 1 (waiting-count climit name))) @@ -350,30 +283,30 @@ (fn [] (th/with-climit [climit {:test-waiting-keys/by-profile {:permits 1 :timeout 3000}}] (let [name (limiter-name :test-waiting-keys/by-profile) - gates {:a (promise) :b (promise)} + gates {"a" (promise) "b" (promise)} holders (into {} (map (fn [k] (let [entered (promise)] [k {:entered entered - :future (future (climit/invoke! (limiter-cfg climit :test-waiting-keys/by-profile k) - (fn [_ _] - (deliver entered true) - @(gates k) - :held) - {}))}]))) - [:a :b])] + :future (future (invoke! (limiter-cfg climit :test-waiting-keys/by-profile k) + (fn [_ _] + (deliver entered true) + @(gates k) + :held) + {}))}]))) + ["a" "b"])] - (doseq [k [:a :b]] + (doseq [k ["a" "b"]] (t/is (true? (deref (:entered (holders k)) 5000 false)))) ;; one call per key, waiting behind its own holder (let [waiters (into {} (map (fn [k] - [k (future (climit/invoke! (limiter-cfg climit :test-waiting-keys/by-profile k) - (fn [_ _] :waited) - {}))])) - [:a :b])] + [k (future (invoke! (limiter-cfg climit :test-waiting-keys/by-profile k) + (fn [_ _] :waited) + {}))])) + ["a" "b"])] (t/is (true? (wait-until #(= 2 (waiting-count climit name)) 5000))) ;; the increment that reaches 2 always publishes 2, whichever waiter @@ -381,9 +314,9 @@ ;; contract, the value is. (t/is (some #(= 2 %) (waiting-value-of name))) - (doseq [k [:a :b]] (deliver (gates k) true)) + (doseq [k ["a" "b"]] (deliver (gates k) true)) - (doseq [k [:a :b]] + (doseq [k ["a" "b"]] (t/is (= :waited (deref (waiters k) 5000 :timeout))) (t/is (= :held (deref (:future (holders k)) 5000 :timeout)))) @@ -413,17 +346,17 @@ cfg (limiter-cfg climit :test-rejected/by-profile "a") entered (promise) gate (promise) - holder (future (climit/invoke! cfg - (fn [_ _] (deliver entered true) @gate :held) - {}))] + holder (future (invoke! cfg + (fn [_ _] (deliver entered true) @gate :held) + {}))] (t/is (true? (deref entered 5000 false))) (reset! calls []) ;; the queue holds one waiter, so the next call is rejected right away - (let [queued (future (failed-type #(climit/invoke! cfg (fn [_ _] :queued) {})))] + (let [queued (future (failed-type #(invoke! cfg (fn [_ _] :queued) {})))] (t/is (true? (wait-until #(= 1 (waiting-last name)) 5000))) - (t/is (= :concurrency-limit (failed-type #(climit/invoke! cfg (fn [_ _] :extra) {})))) + (t/is (= :concurrency-limit (failed-type #(invoke! cfg (fn [_ _] :extra) {})))) (t/is (= [[name "queue-full"]] (rejections))) (deliver gate true) @@ -436,13 +369,13 @@ cfg (limiter-cfg climit :test-timeout/by-profile "a") entered (promise) gate (promise) - holder (future (climit/invoke! cfg - (fn [_ _] (deliver entered true) @gate :held) - {}))] + holder (future (invoke! cfg + (fn [_ _] (deliver entered true) @gate :held) + {}))] (t/is (true? (deref entered 5000 false))) (reset! calls []) - (t/is (= :concurrency-limit (failed-type #(climit/invoke! cfg (fn [_ _] :late) {})))) + (t/is (= :concurrency-limit (failed-type #(invoke! cfg (fn [_ _] :late) {})))) (t/is (= [[name "timeout"]] (rejections))) (deliver gate true) @@ -458,14 +391,14 @@ cfg (limiter-cfg climit :test-rejected-baseline/by-profile "a") entered (promise) gate (promise) - holder (future (climit/invoke! cfg - (fn [_ _] (deliver entered true) @gate :held) - {}))] + holder (future (invoke! cfg + (fn [_ _] (deliver entered true) @gate :held) + {}))] (t/is (true? (deref entered 5000 false))) (reset! calls []) - (t/is (= :concurrency-limit (failed-type #(climit/invoke! cfg (fn [_ _] :extra) {})))) + (t/is (= :concurrency-limit (failed-type #(invoke! cfg (fn [_ _] :extra) {})))) (t/is (= 0 (waiting-count climit name))) (t/is (true? (wait-until #(= 0 (waiting-last name)) 5000))) @@ -482,7 +415,7 @@ (reset! calls []) (t/is (thrown? clojure.lang.ExceptionInfo - (climit/invoke! cfg (fn [_ _] (throw (ex-info "boom" {}))) {}))) + (invoke! cfg (fn [_ _] (throw (ex-info "boom" {}))) {}))) (t/is (= 0 (waiting-count climit name))) (t/is (true? (wait-until #(= 0 (waiting-last name)) 5000)))))))) @@ -510,7 +443,7 @@ (reset! calls []) (with-redefs [mtx/run! (failing-first-waiting-publish calls)] (t/is (thrown? clojure.lang.ExceptionInfo - (climit/invoke! cfg (fn [_ _] :never) {})))) + (invoke! cfg (fn [_ _] :never) {})))) ;; the count came back down instead of staying up for good (t/is (= 0 (waiting-count climit name))) @@ -520,11 +453,11 @@ ;; run inline instead of waiting for the permit another thread holds. (let [entered (promise) gate (promise) - holder (future (climit/invoke! cfg - (fn [_ _] (deliver entered true) @gate :held) - {}))] + holder (future (invoke! cfg + (fn [_ _] (deliver entered true) @gate :held) + {}))] (t/is (true? (deref entered 5000 false))) - (t/is (= :concurrency-limit (failed-type #(climit/invoke! cfg (fn [_ _] :inline) {})))) + (t/is (= :concurrency-limit (failed-type #(invoke! cfg (fn [_ _] :inline) {})))) (deliver gate true) (t/is (= :held (deref holder 5000 :timeout))))))) @@ -549,13 +482,13 @@ (let [cfg (limiter-cfg climit :test-registry/by-profile "a") entered (promise) gate (promise) - holder (future (climit/invoke! cfg - (fn [_ _] (deliver entered true) @gate :held) - {}))] + holder (future (invoke! cfg + (fn [_ _] (deliver entered true) @gate :held) + {}))] (t/is (true? (deref entered 5000 false))) - (let [waiter (future (climit/invoke! cfg (fn [_ _] :waited) {}))] + (let [waiter (future (invoke! cfg (fn [_ _] :waited) {}))] (t/is (true? (wait-until #(= 1.0 (gauge-value metrics name)) 5000))) (deliver gate true) (t/is (= :waited (deref waiter 5000 :timeout))) @@ -592,20 +525,20 @@ cfg (limiter-cfg climit :test-log/by-profile "a") entered (promise) gate (promise) - holder (future (climit/invoke! cfg - (fn [_ _] (deliver entered true) @gate :held) - {}))] + holder (future (invoke! cfg + (fn [_ _] (deliver entered true) @gate :held) + {}))] (t/is (true? (deref entered 5000 false))) (reset! calls []) - (let [q1 (future (climit/invoke! cfg (fn [_ _] :q1) {})) - q2 (future (climit/invoke! cfg (fn [_ _] :q2) {}))] + (let [q1 (future (invoke! cfg (fn [_ _] :q1) {})) + q2 (future (invoke! cfg (fn [_ _] :q2) {}))] ;; the two of them fill the queue, so the next one is rejected (t/is (true? (wait-until #(= 2 (waiting-count climit name)) 5000))) (t/is (some #(= 2 %) (waiting-value-of name))) - (t/is (= :concurrency-limit (failed-type #(climit/invoke! cfg (fn [_ _] :x) {})))) + (t/is (= :concurrency-limit (failed-type #(invoke! cfg (fn [_ _] :x) {})))) (let [props (::l/props (rejected-record records "queue-full"))] (t/is (= 1 (get props :permits))) @@ -630,14 +563,14 @@ gate (promise) capture (fn [_ action _req-id _stats _elapsed code reason] (swap! lines conj [action code reason])) - holder (future (climit/invoke! cfg - (fn [_ _] (deliver entered true) @gate :held) - {}))] + holder (future (invoke! cfg + (fn [_ _] (deliver entered true) @gate :held) + {}))] (t/is (true? (deref entered 5000 false))) (with-redefs [climit/log capture] - (t/is (= :concurrency-limit (failed-type #(climit/invoke! cfg (fn [_ _] :late) {})))) + (t/is (= :concurrency-limit (failed-type #(invoke! cfg (fn [_ _] :late) {})))) (t/is (= [["rejected" :capacity-limit-reached "timeout"]] (filterv #(= "rejected" (first %)) @lines)))) diff --git a/backend/test/backend_tests/helpers.clj b/backend/test/backend_tests/helpers.clj index 8aa3b684fc..58b36244ea 100644 --- a/backend/test/backend_tests/helpers.clj +++ b/backend/test/backend_tests/helpers.clj @@ -577,13 +577,19 @@ (defn init-climit "Build a concurrency limiter component from an inline config map. The optional second argument is the metrics instance to publish to (a - `metrics-stub` by default, where every `run!` is a no-op)." + `metrics-stub` by default, where every `run!` is a no-op). + + The root limiters are added with permits that no test reaches: every + request goes through them, and a test that does not list them would log a + warning per request." ([config] (init-climit config (metrics-stub))) ([config metrics] (let [file (doto (java.io.File/createTempFile "climit" ".edn") (.deleteOnExit))] - (spit file (pr-str config)) + (spit file (pr-str (merge {:root/global {:permits 1000 :timeout 3000} + :root/by-profile {:permits 1000 :timeout 3000}} + config))) (ig/init-key ::rpc/climit {::climit/enabled true ::climit/config (.getAbsolutePath file) diff --git a/backend/test/backend_tests/rpc_binfile_test.clj b/backend/test/backend_tests/rpc_binfile_test.clj index e684d5b7e0..f15a261040 100644 --- a/backend/test/backend_tests/rpc_binfile_test.clj +++ b/backend/test/backend_tests/rpc_binfile_test.clj @@ -175,14 +175,14 @@ (defn- import-file [profile path] - (let [out (th/command! {::th/type :import-binfile - ::rpc/profile-id (:id profile) - :project-id (:default-project-id profile) - :name "imported" - :file {:filename "imported.penpot" - :size (fs/size path) - :path path - :mtype "application/zip"}})] + (let [out (th/multipart-command! {::th/type :import-binfile + ::rpc/profile-id (:id profile) + :project-id (:default-project-id profile) + :name "imported" + :file {:filename "imported.penpot" + :size (fs/size path) + :path path + :mtype "application/zip"}})] (t/is (nil? (:error out))) (th/consume-sse (:result out)))) diff --git a/backend/test/backend_tests/rpc_climit_test.clj b/backend/test/backend_tests/rpc_climit_test.clj index b5b8e853d8..aa058f03d2 100644 --- a/backend/test/backend_tests/rpc_climit_test.clj +++ b/backend/test/backend_tests/rpc_climit_test.clj @@ -6,13 +6,22 @@ (ns backend-tests.rpc-climit-test (:require + [app.rpc :as-alias rpc] [app.rpc.climit :as-alias 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])) (t/use-fixtures :once th/state-init) +(defn- command-limits + "The `::climit/id` declaration of the RPC command named `name`." + [name] + (let [[mdata _] (get (:app.rpc/methods th/*system*) (keyword name))] + (::climit/id mdata))) + (defn- method-limit-declarations "The `::climit/id` declarations of every RPC command, as the running system wraps them." @@ -24,11 +33,87 @@ (when-let [limits (::climit/id mdata)] [mdata limits]))))) +(defn- declared-ids + "The limit ids a command declares, in the order it declares them." + [limits] + (cond + (keyword? limits) + [limits] + + (vector? (first limits)) + (mapv first limits) + + :else + [(first limits)])) + +(defn- limit-order-edges + "The pairs of limit ids the commands declare, as edges: the first one of the + pair is taken before the second one." + [declarations] + (reduce (fn [edges [mdata limits]] + (let [ids (declared-ids limits) + size (count ids)] + (into edges + (for [i (range size) + j (range (inc i) size)] + [(nth ids i) (nth ids j)])))) + #{} + declarations)) + +(defn- limit-order-cycle + "The limit ids whose relative order the commands disagree about: the ones + left when the relation cannot be sorted. + + Two commands that take the same limiters in opposite order leave two paths + of one profile holding them in opposite order, and the threads park for + good. A ring of three does the same, so the relation is checked as a whole + and not pair by pair." + [edges] + (let [nodes (into #{} (mapcat identity) edges) + outgoing (reduce (fn [acc [a b]] (update acc a (fnil conj #{}) b)) {} edges) + indegree (reduce (fn [acc [_ b]] (update acc b (fnil inc 0))) {} edges)] + (loop [pending nodes + degrees indegree] + (let [ready (filterv #(zero? (get degrees % 0)) pending)] + (if (empty? ready) + (seq pending) + (recur (apply disj pending ready) + (reduce (fn [degrees node] + (reduce (fn [degrees next] + (update degrees next (fnil dec 0))) + degrees + (get outgoing node))) + degrees + ready))))))) + +(t/deftest the-media-object-from-url-command-declares-its-own-limits + ;; The command used to enter climit from inside its handler, which forced + ;; every path to take the root limiters before the command ones. It declares + ;; its limits now, so there is a single place where they are read. + (t/is (= [[:process-image/by-profile ::rpc/profile-id] + [:process-image/global]] + (command-limits "create-file-media-object-from-url")))) + +(t/deftest the-update-profile-photo-command-declares-its-own-limits + ;; The command used to enter climit from inside its handler, which forced + ;; every path to take the root limiters before the command ones. It declares + ;; its limits now, so there is a single place where they are read. + (t/is (= [[:process-image/by-profile ::rpc/profile-id] + [:process-image/global]] + (command-limits "update-profile-photo")))) + +(t/deftest the-update-team-photo-command-declares-its-own-limits + ;; The command generates the thumbnail of a team through `upload-photo`, + ;; which used to enter climit on its own. It declares the limits now, so the + ;; image processing of a team is bounded like the one of a profile. + (t/is (= [[:process-image/by-profile ::rpc/profile-id] + [:process-image/global]] + (command-limits "update-team-photo")))) + (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. + ;; declared is acquired first. 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 @@ -49,3 +134,46 @@ (t/is (< idx gidx) (str where " must declare " id " before " (keyword (namespace id) "global"))))))))) + +(t/deftest every-command-takes-the-shared-limits-in-the-same-order + ;; Two commands that share a limit and take it in opposite order can hold + ;; the same limiters in opposite order and park for good. The order is free + ;; now that no command enters climit from inside its handler, so this test + ;; is what keeps the commands consistent with each other. + (let [declarations (method-limit-declarations) + edges (limit-order-edges declarations) + cycle (limit-order-cycle edges)] + + ;; the system wraps every command, so a short list means the scan went stale + (t/is (<= 10 (count declarations))) + (t/is (seq edges)) + (t/is (nil? cycle) + (str "the commands take these limits in different orders: " + (pr-str cycle))))) + +(t/deftest a-ring-of-three-limits-is-a-cycle + ;; A ring of three limits does the same as a pair in opposite order: three + ;; paths hold the three limiters and wait for each other. The relation is + ;; checked as a whole for that reason, and not pair by pair. + (t/is (= #{:a :b :c} + (set (limit-order-cycle #{[:a :b] [:b :c] [:c :a]})))) + (t/is (nil? (limit-order-cycle #{[:a :b] [:b :c]})))) + +(t/deftest every-declared-limit-has-an-entry-in-the-shipped-config + ;; `climit/wrap` logs a warning and runs the handler with no limiter at all + ;; when the config has no entry for a declared id, so the command runs on + ;; every thread that asks for it. The production file is maintained by hand + ;; outside the repository and mirrors this one, so a command declaring a limit + ;; the file does not list is the shape of an unbounded command in production. + (let [path (io/resource "climit.edn") + config (edn/read-string (slurp path)) + declared (into #{} + (mapcat (fn [[_ limits]] (declared-ids limits))) + (method-limit-declarations))] + + ;; the system wraps every command, so an empty set means the scan went stale + (t/is (seq declared)) + + (doseq [id declared] + (t/is (some? (get config id)) + (str id " is declared by a command but has no entry in climit.edn")))))