mirror of
https://github.com/penpot/penpot.git
synced 2026-11-05 09:30:38 +00:00
* ✨ Stabilize fragmented file data across snapshots and GC Snapshots, automatic and user, now store pointer-based rows that share fragments with the live file instead of duplicating page content. `get-file-snapshot` resolves those pointers on the server so version previews keep rendering plain data, and `restore!` reuses the shared rows. Export still materializes (the package carries no fragments) and import regenerates them. `file-gc` treats the main file and every unpurged snapshot as liveness roots, reads the file unrealized, and reattaches the original pointer wherever the cleaned value is unchanged, so a pass over an untouched file writes no fragments and a pass over an edited file keeps the ids of the pages it did not touch. Snapshot creation and the restore backup take the file row `FOR SHARE`, which serializes them with the pass and closes the race that let a new snapshot reference fragments the pass was about to mark. All backend pointer bindings go through the `fdata/with-pointers`, `with-tracked`, `with-tracked-pointers` and `with-untracked` macros. The loader reads fragments a previous pass marked until `objects-gc` purges them, and the frontend fails loudly with retry instead of dropping an unresolved fragment. Review fixes in this branch: the automatic snapshot attaches before pending migrations, `realize-pointers` is the single verb, `reattach-unchanged` takes the identity fast path, the restore guard takes its revn from the backup read, and `PointerMap.withMeta` generates a fresh id when metadata changes. Tests cover both snapshot writers (red without the lock), snapshot sharing and pin release through the real delete and purge paths, reattach against orphan marking, preview after offload, the pointer binding contracts and the with-meta id change. Closes #12410 AI-assisted-by: muse-spark-1.3-contributor AI-assisted-by: deepseek-v4.1-flash ♻️ Pin no-churn behavior for untouched pages Pin the behavior the churn verification measured: a save keeps the fragment id of every page it does not touch, with a page of eleven keys (PersistentHashMap), a 46-shape objects map and a nine-page index (hash map). Also pin the mixed case: a data-level objects-map flip rewrites the old pointer pages once and the next save keeps the new ids. The memory records the identity rule behind it: re-assoc with the same value returns the same map in array and hash maps alike. Closes #12410 AI-assisted-by: deepseek-v4.1-flash ⚡ Stop loading untouched pages when enabling objects-map `enable-objects-map` wrapped every page on every save, and reading a page's objects forced its PointerMap to load the fragment. A save of one page in a seven-page file therefore made seven fragment reads. Skip unloaded pointers: their fragment was already wrapped by the encode-file that wrote it, so loading it only to wrap it again is a no-op. The save now loads only the pages it edits. A page with a plain fragment (a data-level feature flip) keeps its plain fragment until it is first loaded or edited; a migration that enables the feature must rewrite the data itself. Cover the change with load-count tests for save and restore, update the flip test to the new contract, and drop the stale restore comment. Closes #12410 AI-assisted-by: deepseek-v4.1-flash ⚡ Keep page fragment ids through pending migrations Applying a pending migration on read realized the file, ran the migration and persisted it again; the persist wrapped every plain page in a new PointerMap, so every page got a new fragment id even when the migration did not change it. That leaves the fragment cache cold and rewrites every fragment on the first read after a deploy. Capture the original pointers before realizing, reattach the unchanged positions after migrating (same value comparison file-gc already uses), persist the reattached file and return the realized one: the read contract is plain data. Read-only connections skip the persist and the reattach. A migration that adds objects-map skips the reattach so the pages get wrapped. Move the three helpers to fdata so file-gc and the migration path share them, and cover the path with a real 0031 migration test: the unchanged page keeps its id, the changed page gets a new one, the caller gets realized data and a second read does not migrate again. Closes #12410 AI-assisted-by: deepseek-v4.1-flash * 🐛 Realize library pointers with each library's own loader Fragment rows are keyed by file id, and every caller binds the fragment loader of the file it works on. get-resolved-file-libraries returned libraries with their pages still as pointers, so the first read of a library page under fdata/pointer-map asked the consuming file for the fragment and failed with :fragment-not-found. That broke migrations and file validation for every file using a pointer-mapped library. The weak-map loader now realizes each library's pointers with the library's own id. The test fails without the change. AI-assisted-by: claude-opus-5-5 Signed-off-by: Álvaro Tejero Cantero <alvorithm@teje.ro> * 🐛 Restore the pointer-map require the repair path still needs The branch dropped `app.util.pointer-map` from the namespace while the auto-repair path (added to develop before this branch) still rebinds `pmap/*tracked*` around validation and repair in `repair-preexisting-damage` and `process-changes-and-validate`. The namespace did not compile, so the whole backend failed to load. Those toggles do not fit the fdata macros: `with-untracked` binds nil and `with-tracked` creates a fresh atom, while the repair path has to turn tracking off and restore the existing atom (and read it before the repair starts). Restore the require and keep the bindings explicit. Also pin the feature-off path with a test: with `fdata/pointer-map` off (the default), a save creates no fragments, file-gc writes nothing and the snapshot preview still returns plain data. Drop a stray `\n` char literal left at the end of one deftest. Update the memories with the exception to the macro rule, the snapshot FOR SHARE locking and the pre-migration automatic snapshot. Closes #12410 AI-assisted-by: deepseek-v4.1-flash * 🐛 Validate debug files with their own pointer loader The `/dbg/actions` file validation passed the file id where `get-resolved-file-libraries` expects the file map, so the library map was seeded with a nil key. It also validated a file read without realizing it, so the first page lookup of a pointer-mapped file raised "load is not supported when *load-fn* is not bind" and the endpoint could not validate exactly the files the feature touches. Read the file realized (validation walks every page and component anyway) and build the library map from the file. The library loader already resolves each library's pointers with its own id, so the realized file and its libraries validate as plain data. Covered by a handler-level test that fails before the change. AI-assisted-by: deepseek-v4.1-flash * ♻️ Bind the snapshot loader through the fdata helper `reduce-snapshots` hand-rolled the same `binding` that `fdata/with-pointers` builds (it also loaded the function's own pointer loader before entering the reduction). Use the macro so the rule holds there too and the namespace no longer needs `app.util.pointer-map`; the loader is identical (same cfg and file-id) and snapshot walking behaves the same. AI-assisted-by: deepseek-v4.1-flash * 📚 Note the pointer-binding exceptions and snapshot ordering Record what the feature review settled: - Two direct `binding`s stay on purpose: the fine-grained `*tracked*` toggles of `app.rpc.commands.files-update` (validation/repair turn tracking off and restore the existing atom) and `binfile.common/encode-file` (fresh tracked set inheriting the current one). Atom-scoped macros would only be worth it if the set grows. - The automatic snapshot is taken before the pending migration, and the pre-migration window is narrow: the `get-file` RPC migrates on read and its ETag hashes the available migrations, so a post-deploy read cannot be a 304 from the pre-deploy state. The old-format snapshot only appears when the save is the first request to touch the file after the deploy, and `get-file-snapshot` previews it as stored. AI-assisted-by: deepseek-v4.1-flash * 🐛 Apply pending migrations when previewing a version Automatic snapshots are attached before the save applies pending file migrations, so a version can store the old format. The preview RPC handed that data to the UI as stored, and the viewport renders it with the current code. Realize the snapshot with the live file id (fragments are keyed by file, not by snapshot), normalize its migrations to a set (the decoder keeps the pgarray as a vector, and `set/difference` would compare indexes) and apply the pending migrations with the pure `fmg/migrate-file` and the live libraries. Nothing is persisted: the stored version keeps its data and restoring it re-migrates on the next read. The response now carries the snapshot's migrations. Use `:features` the way `get-file` does for `fdata/objects-map`: a client that does not declare it gets plain objects. The historical feature set is not checked against the enabled one on purpose: a version may predate a feature change. Cover the preview of a snapshot with a pending migration and the objects realization; both pin that the stored row is untouched. Update the file-data memory with the preview contract and the vector/set trap. AI-assisted-by: deepseek-v4.1-flash * 🐛 Decode snapshot migrations as an ordered set `file-snapshots/decode-snapshot` handed the `file_change.migrations` pgarray as a vector. `set/difference` falls back to index lookups when the first collection is shorter than the second, so the pending set was right only by accident: a stored subset happens to work, while a stored set larger than `available-migrations` (a snapshot taken by a newer build) marks every available migration as pending and re-applies it to already migrated data. Decode it as a `d/ordered-set`, the type the rest of the code and `schema:file` expect. This also covers `restore!` and the snapshot list, and it lets `get-file-snapshot` drop the `(into #{} ...)` normalization it was carrying. Re-wrap the union in `fmg/migrate` so the invariant holds even when a caller hands a plain set. A snapshot with NULL `migrations` predates SQL migration 0140 and keeps the format version of its time. Pin that legacy path: the preview regenerates the applied set from that version, migrates in memory and leaves the stored row untouched. Update the file-data memory with the invariant and the legacy path. AI-assisted-by: deepseek-v4.1-flash * 🐛 Fix flaky waiting-gauge assertion in climit test The throwable-while-waiting test asserted the exact list of waiting-gauge publications, but the component sampler republishes the same count when its first tick lands inside the recorded window under load. Assert only the last value, which is the contract the helper documents. AI-assisted-by: muse-spark-1.3 --------- Signed-off-by: Álvaro Tejero Cantero <alvorithm@teje.ro> Co-authored-by: Álvaro Tejero Cantero <alvorithm@teje.ro>
629 lines
27 KiB
Clojure
629 lines
27 KiB
Clojure
;; 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.main :as main]
|
|
[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]
|
|
[integrant.core :as ig]))
|
|
|
|
(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 val]}]
|
|
(swap! calls conj {:id id :labels (vec labels) :val val}))]
|
|
(f)))
|
|
|
|
(defn- acquired-in-order
|
|
"Limiter names acquired so far, in acquisition order."
|
|
[]
|
|
(->> @calls
|
|
(filter #(= :rpc-climit-timing (:id %)))
|
|
(mapcat :labels)
|
|
(vec)))
|
|
|
|
(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."
|
|
[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-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]
|
|
(let [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 (= ["process-image/by-profile"
|
|
"process-image/global"
|
|
"root/by-profile"
|
|
"root/global"]
|
|
(acquired-in-order))))))))
|
|
|
|
(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 [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))))))))
|
|
|
|
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
;; 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. 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"))
|
|
(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
|
|
;; 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/with-climit [_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)))))
|
|
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
;; WAITING COUNTER
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
|
|
(defn- waiting-value-of
|
|
"Values published to the waiting gauge for one limiter name, in order."
|
|
[name]
|
|
(->> @calls
|
|
(filter #(= :rpc-climit-waiting (:id %)))
|
|
(filter #(= name (first (:labels %))))
|
|
(map :val)
|
|
(vec)))
|
|
|
|
(defn- waiting-last
|
|
"Last value published to the waiting gauge for one limiter name. How many
|
|
publications land inside a window is not part of the contract, the value is,
|
|
so this is what the assertions read."
|
|
[name]
|
|
(last (waiting-value-of name)))
|
|
|
|
(defn- waiting-count
|
|
"Current waiting count of a limiter name, read from its component."
|
|
[climit name]
|
|
(.get ^java.util.concurrent.atomic.AtomicLong
|
|
(#'climit/waiting-counter (:counters climit) name)))
|
|
|
|
(defn- wait-until
|
|
"Wait until `pred` holds, or give up."
|
|
[pred timeout]
|
|
(let [deadline (+ (System/currentTimeMillis) timeout)]
|
|
(loop []
|
|
(cond
|
|
(pred) true
|
|
(> (System/currentTimeMillis) deadline) false
|
|
:else (do (Thread/sleep 10) (recur))))))
|
|
|
|
(defn- limiter-name
|
|
"The name climit publishes for a limiter id: the keyword without the colon."
|
|
[id]
|
|
(subs (str id) 1))
|
|
|
|
(defn- limiter-cfg
|
|
[climit name key]
|
|
{::rpc/climit climit
|
|
::climit/id [[name key]]})
|
|
|
|
(t/deftest waiting-gauge-follows-the-queue
|
|
;; The value is the number of calls waiting for a permit: up when a call
|
|
;; starts waiting, down when the wait ends. The gauges that sampled the
|
|
;; limiter stats could not show this, because the value of whichever limiter
|
|
;; of the name ran last was the one that survived.
|
|
(record-metrics
|
|
(fn []
|
|
(th/with-climit [climit {:test-waiting/by-profile {:permits 1 :timeout 3000}}]
|
|
(let [name (limiter-name :test-waiting/by-profile)
|
|
entered (promise)
|
|
gate (promise)
|
|
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 (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)))
|
|
|
|
(deliver gate true)
|
|
(t/is (= :waited (deref waiter 5000 :timeout)))
|
|
(t/is (= :held (deref holder 5000 :timeout)))
|
|
(t/is (true? (wait-until #(= 0 (waiting-last name)) 5000)))
|
|
(t/is (= 0 (waiting-count climit name)))))))))
|
|
|
|
(t/deftest waiting-gauge-sums-every-key-of-a-name
|
|
;; Two profiles queueing on the same limiter name add up. That is the shape
|
|
;; that stayed invisible in production: one series per name, counting the
|
|
;; entries and the exits, instead of one value per limiter overwritten by the
|
|
;; next one.
|
|
(record-metrics
|
|
(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)}
|
|
holders
|
|
(into {}
|
|
(map (fn [k]
|
|
(let [entered (promise)]
|
|
[k {:entered entered
|
|
: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"]]
|
|
(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 (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
|
|
;; does it last: the order of two concurrent publications is not the
|
|
;; contract, the value is.
|
|
(t/is (some #(= 2 %) (waiting-value-of name)))
|
|
|
|
(doseq [k ["a" "b"]] (deliver (gates k) true))
|
|
|
|
(doseq [k ["a" "b"]]
|
|
(t/is (= :waited (deref (waiters k) 5000 :timeout)))
|
|
(t/is (= :held (deref (:future (holders k)) 5000 :timeout))))
|
|
|
|
(t/is (true? (wait-until #(= 0 (waiting-count climit name)) 5000)))
|
|
(t/is (some #(= 0 %) (waiting-value-of name)))))))))
|
|
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
;; REJECTIONS
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
|
|
(defn- rejections
|
|
"Rejections published, as [name reason] pairs."
|
|
[]
|
|
(->> @calls
|
|
(filter #(= :rpc-climit-rejected (:id %)))
|
|
(map (fn [call] [(first (:labels call)) (second (:labels call))]))
|
|
(vec)))
|
|
|
|
(t/deftest rejection-counter-tells-why
|
|
;; Rejections had no metric: the panel counted them from the logs. Now a call
|
|
;; rejected because the queue is full and a call rejected because it waited
|
|
;; too long are counted apart.
|
|
(record-metrics
|
|
(fn []
|
|
(th/with-climit [climit {:test-rejected/by-profile {:permits 1 :queue 1 :timeout 60000}}]
|
|
(let [name (limiter-name :test-rejected/by-profile)
|
|
cfg (limiter-cfg climit :test-rejected/by-profile "a")
|
|
entered (promise)
|
|
gate (promise)
|
|
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 #(invoke! cfg (fn [_ _] :queued) {})))]
|
|
(t/is (true? (wait-until #(= 1 (waiting-last name)) 5000)))
|
|
(t/is (= :concurrency-limit (failed-type #(invoke! cfg (fn [_ _] :extra) {}))))
|
|
(t/is (= [[name "queue-full"]] (rejections)))
|
|
|
|
(deliver gate true)
|
|
(t/is (= :held (deref holder 5000 :timeout)))
|
|
(t/is (= :queued (deref queued 5000 :timeout))))))
|
|
|
|
;; a wait that times out is counted as such, not as a full queue
|
|
(th/with-climit [climit {:test-timeout/by-profile {:permits 1 :timeout 300}}]
|
|
(let [name (limiter-name :test-timeout/by-profile)
|
|
cfg (limiter-cfg climit :test-timeout/by-profile "a")
|
|
entered (promise)
|
|
gate (promise)
|
|
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 #(invoke! cfg (fn [_ _] :late) {}))))
|
|
(t/is (= [[name "timeout"]] (rejections)))
|
|
|
|
(deliver gate true)
|
|
(t/is (= :held (deref holder 5000 :timeout))))))))
|
|
|
|
(t/deftest the-waiting-count-comes-back-after-a-rejection
|
|
;; A rejected call increments and decrements exactly once: a limiter that
|
|
;; rejects under pressure must not drift the panel up.
|
|
(record-metrics
|
|
(fn []
|
|
(th/with-climit [climit {:test-rejected-baseline/by-profile {:permits 1 :queue 0 :timeout 3000}}]
|
|
(let [name (limiter-name :test-rejected-baseline/by-profile)
|
|
cfg (limiter-cfg climit :test-rejected-baseline/by-profile "a")
|
|
entered (promise)
|
|
gate (promise)
|
|
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 #(invoke! cfg (fn [_ _] :extra) {}))))
|
|
(t/is (= 0 (waiting-count climit name)))
|
|
(t/is (true? (wait-until #(= 0 (waiting-last name)) 5000)))
|
|
|
|
(deliver gate true)
|
|
(t/is (= :held (deref holder 5000 :timeout))))))))
|
|
|
|
(t/deftest the-waiting-count-comes-back-when-the-handler-throws
|
|
;; A handler that throws still leaves the count where it was.
|
|
(record-metrics
|
|
(fn []
|
|
(th/with-climit [climit {:test-handler-throws/by-profile {:permits 1 :timeout 3000}}]
|
|
(let [name (limiter-name :test-handler-throws/by-profile)
|
|
cfg (limiter-cfg climit :test-handler-throws/by-profile "a")]
|
|
|
|
(reset! calls [])
|
|
(t/is (thrown? clojure.lang.ExceptionInfo
|
|
(invoke! cfg (fn [_ _] (throw (ex-info "boom" {}))) {})))
|
|
|
|
(t/is (= 0 (waiting-count climit name)))
|
|
(t/is (true? (wait-until #(= 0 (waiting-last name)) 5000))))))))
|
|
|
|
(defn- failing-first-waiting-publish
|
|
"A `mtx/run!` replacement that fails the first waiting publication and records
|
|
the rest, so the decrement of the unwinding path still lands."
|
|
[calls]
|
|
(let [pending (atom true)]
|
|
(fn [_ & {:keys [id labels val]}]
|
|
(if (and @pending (= :rpc-climit-waiting id))
|
|
(do (reset! pending false)
|
|
(throw (ex-info "publishing is down" {})))
|
|
(swap! calls conj {:id id :labels (vec labels) :val val})))))
|
|
|
|
(t/deftest a-throwable-while-waiting-leaves-no-mark-and-no-count
|
|
;; `waiting-inc` sits inside the `try`: a throwable from the publishing path
|
|
;; unwinds through the `finally`, so the thread neither keeps the limiter
|
|
;; marked as held (it would skip that limiter for that key for ever) nor
|
|
;; leaves the count up.
|
|
(th/with-climit [climit {:test-throw-waiting/by-profile {:permits 1 :timeout 3000}}]
|
|
(let [name (limiter-name :test-throw-waiting/by-profile)
|
|
cfg (limiter-cfg climit :test-throw-waiting/by-profile "a")]
|
|
|
|
(reset! calls [])
|
|
(with-redefs [mtx/run! (failing-first-waiting-publish calls)]
|
|
(t/is (thrown? clojure.lang.ExceptionInfo
|
|
(invoke! cfg (fn [_ _] :never) {}))))
|
|
|
|
;; the count came back down instead of staying up for good. Only the
|
|
;; last publication is asserted, not the exact list: the sampler
|
|
;; republishes the same count when its first tick lands inside the
|
|
;; window.
|
|
(t/is (= 0 (waiting-count climit name)))
|
|
(t/is (= 0 (waiting-last name)))
|
|
|
|
;; and the thread dropped the mark: with it left behind, this call would
|
|
;; run inline instead of waiting for the permit another thread holds.
|
|
(let [entered (promise)
|
|
gate (promise)
|
|
holder (future (invoke! cfg
|
|
(fn [_ _] (deliver entered true) @gate :held)
|
|
{}))]
|
|
(t/is (true? (deref entered 5000 false)))
|
|
(t/is (= :concurrency-limit (failed-type #(invoke! cfg (fn [_ _] :inline) {}))))
|
|
(deliver gate true)
|
|
(t/is (= :held (deref holder 5000 :timeout)))))))
|
|
|
|
(defn- gauge-value
|
|
"Current value of the waiting gauge for one limiter name, read from a real
|
|
prometheus collector."
|
|
[metrics name]
|
|
(let [collector (:app.metrics.definition/instance
|
|
(mtx/get-collector metrics :rpc-climit-waiting))]
|
|
(.get ^io.prometheus.client.Gauge$Child
|
|
(.labels ^io.prometheus.client.Gauge collector (into-array String [name])))))
|
|
|
|
(t/deftest the-waiting-gauge-reaches-a-real-registry
|
|
;; The recorded calls prove the arguments, not that a collector with that name
|
|
;; and those labels exists: a typo in the metric id or a wrong label arity
|
|
;; would only show up in production. Here the value is read back from a real
|
|
;; prometheus registry.
|
|
(let [metrics (ig/init-key :app.metrics/metrics {:default main/default-metrics})
|
|
name (limiter-name :test-registry/by-profile)]
|
|
|
|
(th/with-climit [climit {:test-registry/by-profile {:permits 1 :timeout 3000}} metrics]
|
|
(let [cfg (limiter-cfg climit :test-registry/by-profile "a")
|
|
entered (promise)
|
|
gate (promise)
|
|
holder (future (invoke! cfg
|
|
(fn [_ _] (deliver entered true) @gate :held)
|
|
{}))]
|
|
|
|
(t/is (true? (deref entered 5000 false)))
|
|
|
|
(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)))
|
|
(t/is (= :held (deref holder 5000 :timeout)))
|
|
(t/is (true? (wait-until #(= 0.0 (gauge-value metrics name)) 5000))))))))
|
|
|
|
(defn- rejected-record
|
|
"The first captured log record that reports a rejection with `reason`."
|
|
[records reason]
|
|
(let [deadline (+ (System/currentTimeMillis) 5000)]
|
|
(loop []
|
|
(let [record (first (filter (fn [record]
|
|
(let [props (::l/props record)]
|
|
(and (= "rejected" (get props :hint))
|
|
(= reason (get props :reason)))))
|
|
@records))]
|
|
(cond
|
|
record record
|
|
(> (System/currentTimeMillis) deadline) nil
|
|
:else (do (Thread/sleep 10) (recur)))))))
|
|
|
|
(t/deftest rejection-logs-say-why-and-how-big-the-limits-are
|
|
;; The line has to say whether the queue was full or the wait timed out, and
|
|
;; how many permits and how much queue the limiter has, or nobody can tell
|
|
;; whether the limiter did its job.
|
|
(let [records (atom [])]
|
|
(add-watch l/log-record ::capture
|
|
(fn [_ _ _ record] (swap! records conj record)))
|
|
(try
|
|
(record-metrics
|
|
(fn []
|
|
(th/with-climit [climit {:test-log/by-profile {:permits 1 :queue 2 :timeout 60000}}]
|
|
(let [name (limiter-name :test-log/by-profile)
|
|
cfg (limiter-cfg climit :test-log/by-profile "a")
|
|
entered (promise)
|
|
gate (promise)
|
|
holder (future (invoke! cfg
|
|
(fn [_ _] (deliver entered true) @gate :held)
|
|
{}))]
|
|
|
|
(t/is (true? (deref entered 5000 false)))
|
|
(reset! calls [])
|
|
|
|
(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 #(invoke! cfg (fn [_ _] :x) {}))))
|
|
|
|
(let [props (::l/props (rejected-record records "queue-full"))]
|
|
(t/is (= 1 (get props :permits)))
|
|
(t/is (= 2 (get props :queue-max)))
|
|
(t/is (= :capacity-limit-reached (get props :code)))
|
|
(t/is (= "queue-full" (get props :reason))))
|
|
|
|
(deliver gate true)
|
|
(t/is (= :held (deref holder 5000 :timeout)))
|
|
(deref q1 5000 nil)
|
|
(deref q2 5000 nil))))
|
|
|
|
;; A wait that times out carries its own reason in the line, next to the
|
|
;; code the limiter raised. The line is emitted at TRACE when nothing
|
|
;; else is left queued behind it (the caller was the only one waiting),
|
|
;; so the call to `log` is captured instead of the record, which the
|
|
;; logger would drop.
|
|
(th/with-climit [climit {:test-log-timeout/by-profile {:permits 1 :timeout 300}}]
|
|
(let [cfg (limiter-cfg climit :test-log-timeout/by-profile "a")
|
|
lines (atom [])
|
|
entered (promise)
|
|
gate (promise)
|
|
capture (fn [_ action _req-id _stats _elapsed code reason]
|
|
(swap! lines conj [action code reason]))
|
|
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 #(invoke! cfg (fn [_ _] :late) {}))))
|
|
(t/is (= [["rejected" :capacity-limit-reached "timeout"]]
|
|
(filterv #(= "rejected" (first %)) @lines))))
|
|
|
|
(deliver gate true)
|
|
(t/is (= :held (deref holder 5000 :timeout)))))))
|
|
|
|
(finally
|
|
(remove-watch l/log-record ::capture)))))
|
|
|
|
(t/deftest republish-writes-the-current-waiting-count
|
|
;; A call that never comes back keeps the count up, but if it is also the
|
|
;; last thing that happened nothing rewrites that series and it goes stale in
|
|
;; the scrape. The republish is what keeps it visible. The counters are built
|
|
;; on their own: with a component, its sampler would also publish the same
|
|
;; count the moment its first tick lands inside the window.
|
|
(let [counters (#'climit/create-waiting-counters)
|
|
name (limiter-name :test-republish/by-profile)
|
|
counter ^java.util.concurrent.atomic.AtomicLong (#'climit/waiting-counter counters name)]
|
|
|
|
(.set counter 3)
|
|
|
|
(record-metrics
|
|
(fn []
|
|
(#'climit/republish-waiting (th/metrics-stub) counters)
|
|
(t/is (= [3] (waiting-value-of name)))))))
|
|
(t/deftest halting-the-component-stops-the-republisher
|
|
;; The republisher is a daemon thread started with the component: halting it
|
|
;; has to stop the thread, or every restart leaves another one running.
|
|
(let [climit (th/init-climit {:test-halt/by-profile {:permits 1 :timeout 3000}})
|
|
sampler ^java.util.concurrent.ScheduledThreadPoolExecutor (:sampler climit)]
|
|
|
|
(t/is (false? (.isShutdown sampler)))
|
|
|
|
(ig/halt-key! ::rpc/climit climit)
|
|
|
|
(t/is (true? (.isShutdown sampler)))))
|
|
|
|
(t/deftest the-climit-metrics-that-remain-are-defined
|
|
;; A typo in a metric id, or a definition that is not there, makes the
|
|
;; publishing silent, which is the failure mode this work removes. And the
|
|
;; gauges that could not tell the truth must be gone.
|
|
(let [metrics (ig/init-key :app.metrics/metrics {:default main/default-metrics})]
|
|
|
|
(t/is (= ["name"]
|
|
(:app.metrics.definition/labels (mtx/get-collector metrics :rpc-climit-waiting))))
|
|
(t/is (= ["name" "reason"]
|
|
(:app.metrics.definition/labels (mtx/get-collector metrics :rpc-climit-rejected))))
|
|
(t/is (some? (mtx/get-collector metrics :rpc-climit-timing)))
|
|
|
|
(t/is (nil? (mtx/get-collector metrics :rpc-climit-queue)))
|
|
(t/is (nil? (mtx/get-collector metrics :rpc-climit-permits)))
|
|
(t/is (nil? (mtx/get-collector metrics :http-worker-busy-threads)))))
|