mirror of
https://github.com/penpot/penpot.git
synced 2026-10-30 14:30:37 +00:00
5eb33add7e
* ✨ Add skip-if-active option to jobs submit Producers that need at most one live job per key can now pass ::jobs/skip-if-active: the submit counts live jobs for the same tenant, name, queue and label in the same transaction and skips the insert (returning nil) when one lives. The option requires an explicit label and documents the same best-effort race caveat as the dedupe it complements. Closes #12265 AI-assisted-by: muse-spark-1.3-contributor * ✨ Record skipped job submissions as a metric Skip-if-active submissions withheld nothing observable: the skip now records penpot_jobs_skipped_total (name, queue) instead of the submitted counter, so dashboards can tell withheld work apart from new work. The insert path is untouched. AI-assisted-by: muse-spark-1.3-contributor * ♻️ Drop jobs counters duplicated by histogram counts The completed and requests counters counted exactly the same events as the _count of their timing histograms, so they go away: terminal totals now come from penpot_jobs_total_ms_count by outcome and request totals from penpot_jobs_request_ms_count by outcome. One event, one family from here on: counters only cover events without latency, and the dispatched vs queue-wait pair stays because pushes and claims are different events. AI-assisted-by: muse-spark-1.3-contributor
316 lines
13 KiB
Clojure
316 lines
13 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
|
|
|
|
(ns backend-tests.jobs-request-test
|
|
(:require
|
|
[app.common.generic-pool :as gpool]
|
|
[app.common.json :as json]
|
|
[app.common.time :as ct]
|
|
[app.config :as cf]
|
|
[app.db :as db]
|
|
[app.jobs :as jobs]
|
|
[app.main :as main]
|
|
[app.metrics :as mtx]
|
|
[app.metrics.definition :as-alias mdef]
|
|
[app.redis :as rds]
|
|
[app.worker :as wrk]
|
|
[backend-tests.helpers :as th]
|
|
[clojure.test :as t]
|
|
[cuerdas.core :as str]
|
|
[integrant.core :as ig])
|
|
(:import
|
|
io.lettuce.core.api.sync.RedisCommands
|
|
java.lang.AutoCloseable))
|
|
|
|
(t/use-fixtures :once th/state-init)
|
|
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
;; HELPERS
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
|
|
(defn- make-metrics []
|
|
(ig/init-key :app.metrics/metrics
|
|
{:default (select-keys main/default-metrics
|
|
[:jobs-request-timing])}))
|
|
|
|
(defn- make-pool
|
|
([] (make-pool (get th/*system* :app.metrics/metrics)))
|
|
([metrics]
|
|
(-> {::jobs/request-pool {::rds/client (get th/*system* :app.redis/client)
|
|
::mtx/metrics metrics}}
|
|
(ig/expand)
|
|
(ig/init)
|
|
(get ::jobs/request-pool))))
|
|
|
|
(defn- queue-key
|
|
[queue]
|
|
(wrk/queue-key (cf/get :tenant) queue))
|
|
|
|
(defn- make-cfg
|
|
[pool]
|
|
{::jobs/request-pool pool})
|
|
|
|
(defn- clear-queues
|
|
[]
|
|
(let [conn (rds/connect {::rds/client (get th/*system* :app.redis/client)
|
|
::mtx/metrics (get th/*system* :app.metrics/metrics)})]
|
|
(try
|
|
(let [cmd (.-cmd conn)
|
|
keys (vec (.keys ^RedisCommands cmd
|
|
(str jobs/reply-key-prefix ":" (cf/get :tenant) ":*")))]
|
|
(when (seq keys)
|
|
(rds/del conn keys))
|
|
(rds/del conn (queue-key :media)))
|
|
(finally
|
|
(rds/close conn)))))
|
|
|
|
(defn- test-fixture [next]
|
|
(th/database-reset
|
|
(fn []
|
|
(clear-queues)
|
|
(next))))
|
|
|
|
(t/use-fixtures :each test-fixture)
|
|
|
|
(defn- with-responder
|
|
"Start a responder on :media that answers the next request with the
|
|
provided response map (usually {:ok ...} or {:error ...}). Returns a
|
|
future that resolves to the decoded request payload [request-id,
|
|
reply-key, cmd, params]."
|
|
[pool response-fn & {:keys [delay-ms timeout-ms]}]
|
|
(let [timeout-ms (or timeout-ms 10000)
|
|
out (promise)]
|
|
(let [fut (future
|
|
(let [conn (rds/connect {::rds/client (get th/*system* :app.redis/client)
|
|
::mtx/metrics (get th/*system* :app.metrics/metrics)})]
|
|
(try
|
|
(when delay-ms
|
|
(Thread/sleep (long delay-ms)))
|
|
(let [key (queue-key :media)
|
|
[_ payload] (rds/blpop conn [key] (ct/duration {:millis timeout-ms}))]
|
|
(if (nil? payload)
|
|
(deliver out ::no-request)
|
|
(let [[request-id reply-key cmd params :as decoded] (json/decode payload)]
|
|
(deliver out decoded)
|
|
(jobs/reply {::rds/pool (::jobs/pool pool)
|
|
::mtx/metrics (::mtx/metrics pool)}
|
|
reply-key
|
|
(response-fn params)))))
|
|
(finally
|
|
(rds/close conn)))))]
|
|
{:future fut
|
|
:payload out})))
|
|
|
|
(defn- get-ttl
|
|
[key]
|
|
(let [conn (rds/connect {::rds/client (get th/*system* :app.redis/client)
|
|
::mtx/metrics (get th/*system* :app.metrics/metrics)})]
|
|
(try
|
|
(let [cmd (.-cmd conn)]
|
|
(.ttl ^RedisCommands cmd key))
|
|
(finally
|
|
(rds/close conn)))))
|
|
|
|
(defn- key-exists?
|
|
[key]
|
|
(let [conn (rds/connect {::rds/client (get th/*system* :app.redis/client)
|
|
::mtx/metrics (get th/*system* :app.metrics/metrics)})]
|
|
(try
|
|
(let [cmd (.-cmd conn)]
|
|
(pos? (.exists ^RedisCommands cmd (make-array String 1 key))))
|
|
(finally
|
|
(rds/close conn)))))
|
|
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
;; TESTS
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
|
|
(defn- histogram-count [metrics id labels]
|
|
(let [definition (get main/default-metrics id)]
|
|
(->> (enumeration-seq
|
|
(.metricFamilySamples ^io.prometheus.client.CollectorRegistry
|
|
(mtx/get-registry metrics)))
|
|
(filter (fn [^io.prometheus.client.Collector$MetricFamilySamples family]
|
|
(= (.-name family) (::mdef/name definition))))
|
|
(mapcat (fn [^io.prometheus.client.Collector$MetricFamilySamples family]
|
|
(.samples family)))
|
|
(filter (fn [^io.prometheus.client.Collector$MetricFamilySamples$Sample sample]
|
|
(and (str/ends-with? (.-name sample) "_count")
|
|
(= labels (vec (.-labelValues sample))))))
|
|
(map (fn [^io.prometheus.client.Collector$MetricFamilySamples$Sample sample]
|
|
(.-value sample)))
|
|
(reduce + 0.0))))
|
|
|
|
(t/deftest request-pool-keeps-its-own-metrics
|
|
(let [metrics-a (make-metrics)
|
|
metrics-b (make-metrics)
|
|
pool-a (make-pool metrics-a)
|
|
pool-b (make-pool metrics-b)]
|
|
(t/is (identical? metrics-a (::mtx/metrics pool-a)))
|
|
(t/is (identical? metrics-b (::mtx/metrics pool-b)))))
|
|
|
|
(t/deftest request-metrics-use-the-final-outcome
|
|
(let [metrics (make-metrics)
|
|
pool (make-pool metrics)
|
|
cfg (make-cfg pool)]
|
|
(with-responder pool (fn [_params] {:ok {:value 1}}))
|
|
(t/is (= {:value 1}
|
|
(jobs/request cfg {::jobs/queue :media
|
|
::jobs/cmd :process
|
|
::jobs/params {}})))
|
|
(t/is (= 1.0 (histogram-count metrics :jobs-request-timing ["replied"])))
|
|
(t/is (thrown-with-msg? Exception #"timeout waiting for the job reply"
|
|
(jobs/request cfg {::jobs/queue :media
|
|
::jobs/cmd :process
|
|
::jobs/params {}
|
|
::jobs/timeout (ct/duration {:millis 100})})))
|
|
(t/is (= 1.0 (histogram-count metrics :jobs-request-timing ["timeout"])))
|
|
(t/is (= 0.0 (histogram-count metrics :jobs-request-timing ["sent"])))))
|
|
|
|
(t/deftest request-returns-the-ok-reply
|
|
(let [pool (make-pool)
|
|
cfg (make-cfg pool)
|
|
payload (with-responder pool (fn [_params] {:ok {:value 42}}))]
|
|
|
|
(let [result (jobs/request cfg {::jobs/queue :media
|
|
::jobs/cmd :process
|
|
::jobs/params {:x 1}})]
|
|
(t/is (= {:value 42} result)))
|
|
|
|
(t/testing "the worker receives the expected payload shape"
|
|
(let [[request-id reply-key cmd params :as decoded] (deref (:payload payload) 10000 ::timeout)]
|
|
(t/is (string? request-id))
|
|
(t/is (string? reply-key))
|
|
(t/is (str/starts-with? reply-key (str jobs/reply-key-prefix ":" (cf/get :tenant) ":")))
|
|
(t/is (= "process" cmd))
|
|
(t/is (= {"x" 1} params)
|
|
(str "params should be plain JSON: " (pr-str decoded)))))))
|
|
|
|
(t/deftest request-propagates-the-error-reply
|
|
(let [pool (make-pool)
|
|
cfg (make-cfg pool)
|
|
_ (with-responder pool
|
|
(fn [_params]
|
|
{:error {:code :processing-failed
|
|
:hint "something went wrong"}}))]
|
|
(t/is (thrown-with-msg? Exception #"something went wrong"
|
|
(jobs/request cfg {::jobs/queue :media
|
|
::jobs/cmd :process
|
|
::jobs/params {:x 1}})))))
|
|
|
|
(t/deftest request-times-out-and-cleans-the-reply-key
|
|
(let [pool (make-pool)
|
|
cfg (make-cfg pool)]
|
|
(t/is (thrown-with-msg? Exception #"timeout waiting for the job reply"
|
|
(jobs/request cfg {::jobs/queue :media
|
|
::jobs/cmd :process
|
|
::jobs/params {:x 1}
|
|
::jobs/timeout (ct/duration {:millis 500})})))
|
|
|
|
(t/testing "no orphan reply key is left behind"
|
|
;; there is no way to know the exact reply key (random request-id);
|
|
;; check no reply keys exist at all after the timeout
|
|
(let [conn (rds/connect {::rds/client (get th/*system* :app.redis/client)
|
|
::mtx/metrics (get th/*system* :app.metrics/metrics)})]
|
|
(try
|
|
(let [cmd (.-cmd conn)
|
|
keys (.keys ^RedisCommands cmd
|
|
(str jobs/reply-key-prefix ":" (cf/get :tenant) ":*"))]
|
|
(t/is (empty? (vec keys))))
|
|
(finally
|
|
(rds/close conn)))))))
|
|
|
|
(t/deftest request-pool-returns-connections-on-success-and-failure
|
|
(let [pool (make-pool)
|
|
cfg (make-cfg pool)
|
|
default-timeout (-> (cf/get-jobs-request-timeout)
|
|
(ct/plus (ct/duration {:seconds 30})))]
|
|
|
|
(t/testing "successful request does not consume the connection"
|
|
(with-responder pool (fn [_params] {:ok {:a 1}}))
|
|
(jobs/request cfg {::jobs/queue :media
|
|
::jobs/cmd :process
|
|
::jobs/params {:x 1}})
|
|
|
|
(t/testing "and the next request works with a pooled connection"
|
|
(with-responder pool (fn [_params] {:ok {:b 2}}))
|
|
(t/is (= {:b 2} (jobs/request cfg {::jobs/queue :media
|
|
::jobs/cmd :process
|
|
::jobs/params {:x 2}})))
|
|
|
|
(t/testing "connection command timeout is restored after use"
|
|
(with-open [^AutoCloseable pooled (gpool/get (::jobs/pool pool))]
|
|
(let [conn @pooled]
|
|
(t/is (= (.toMillis ^java.time.Duration default-timeout)
|
|
(.toMillis ^java.time.Duration (rds/get-timeout conn)))))))))))
|
|
|
|
(t/deftest request-supports-larger-per-call-timeout
|
|
(let [pool (make-pool)
|
|
cfg (make-cfg pool)
|
|
_ (with-responder pool (fn [_params] {:ok {:slow 1}}))]
|
|
|
|
(t/testing "an override bigger than the default works (command timeout raised per call)"
|
|
(t/is (= {:slow 1}
|
|
(jobs/request cfg {::jobs/queue :media
|
|
::jobs/cmd :process
|
|
::jobs/params {:x 1}
|
|
::jobs/timeout (ct/duration {:seconds 130})}))))))
|
|
|
|
(t/deftest request-raises-connection-timeout-to-per-call-override
|
|
(let [pool (make-pool)
|
|
cfg (make-cfg pool)
|
|
seen (atom nil)
|
|
orig rds/set-timeout
|
|
override (ct/duration {:seconds 999})
|
|
_ (with-responder pool (fn [_params] {:ok {:fast 1}}))]
|
|
;; an override above the pool default is honored, not capped
|
|
(with-redefs [rds/set-timeout (fn [conn timeout]
|
|
(reset! seen timeout)
|
|
(orig conn timeout))]
|
|
(t/is (= {:fast 1}
|
|
(jobs/request cfg {::jobs/queue :media
|
|
::jobs/cmd :process
|
|
::jobs/params {:x 1}
|
|
::jobs/timeout override}))))
|
|
(t/is (= (.toMillis ^java.time.Duration (ct/plus override (ct/duration {:seconds 30})))
|
|
(.toMillis ^java.time.Duration @seen)))))
|
|
|
|
(t/deftest late-reply-is-cleaned-by-the-worker-expire
|
|
(let [pool (make-pool)
|
|
cfg (make-cfg pool)
|
|
;; the responder replies 2s after the caller timeout (500ms)
|
|
responder (with-responder pool
|
|
(fn [_params] {:ok {:late 1}})
|
|
:delay-ms 2000)]
|
|
|
|
(t/is (thrown-with-msg? Exception #"timeout waiting for the job reply"
|
|
(jobs/request cfg {::jobs/queue :media
|
|
::jobs/cmd :process
|
|
::jobs/params {:x 1}
|
|
::jobs/timeout (ct/duration {:millis 500})})))
|
|
|
|
(t/testing "the late reply key exists (recreated by the worker) with a short TTL"
|
|
(deref (:future responder) 10000 ::timeout)
|
|
(let [conn (rds/connect {::rds/client (get th/*system* :app.redis/client)
|
|
::mtx/metrics (get th/*system* :app.metrics/metrics)})]
|
|
(try
|
|
(let [cmd (.-cmd conn)
|
|
keys (vec (.keys ^RedisCommands cmd
|
|
(str jobs/reply-key-prefix ":" (cf/get :tenant) ":*")))]
|
|
(t/is (= 1 (count keys)))
|
|
(let [ttl (.ttl ^RedisCommands cmd (first keys))]
|
|
(t/is (pos? ttl))
|
|
(t/is (<= ttl 60))))
|
|
(finally
|
|
(rds/close conn)))))))
|
|
|
|
(t/deftest request-creates-no-job-row
|
|
(let [pool (make-pool)
|
|
cfg (make-cfg pool)
|
|
_ (with-responder pool (fn [_params] {:ok {}}))]
|
|
(jobs/request cfg {::jobs/queue :media ::jobs/cmd :process ::jobs/params {}})
|
|
(t/is (zero? (:cnt (th/db-exec-one! ["SELECT count(*) AS cnt FROM job"]))))))
|