github/penpot
1
0
Fork 0
mirror of https://github.com/penpot/penpot.git synced 2026-10-31 23:10:31 +00:00
Code Issues Packages Projects Releases Wiki Activity
penpot/backend/test/backend_tests/worker_dispatcher_test.clj
Andrey Antukh 5eb33add7e
✨ Add skip-if-active option to jobs submit (#12266) 
* ✨ 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
2026-10-11 18:48:02 +02:00

328 lines
13 KiB
Clojure
Raw Permalink Blame History

;; 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.worker-dispatcher-test
(:require
[app.common.json :as json]
[app.common.time :as ct]
[app.common.uuid :as uuid]
[app.config :as cf]
[app.db :as db]
[app.jobs :as jobs]
[app.main :as main]
[app.metrics :as-alias mtx]
[app.metrics.definition :as-alias mdef]
[app.redis :as rds]
[app.worker :as wrk]
[app.worker.dispatcher :as wdisp]
[backend-tests.helpers :as th]
[clojure.string :as str]
[clojure.test :as t]
[integrant.core :as ig])
(:import
io.prometheus.client.Collector$MetricFamilySamples
io.prometheus.client.Collector$MetricFamilySamples$Sample
io.prometheus.client.Counter
io.prometheus.client.Counter$Child))
(t/use-fixtures :once th/state-init)
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; HELPERS
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
(defn- mk-cfg
[]
{::db/pool th/*pool*
::rds/client (get th/*system* :app.redis/client)
::mtx/metrics (get th/*system* :app.metrics/metrics)
::wrk/tenant (cf/get :tenant)
::batch-size 100
::lease (cf/get-jobs-lease)
::timeout (ct/duration "10s")})
(defn- make-metrics []
(ig/init-key :app.metrics/metrics
{:default (select-keys main/default-metrics
[:jobs-dispatcher-timing])}))
(defn- make-orphan-metrics []
(ig/init-key :app.metrics/metrics
{:default (select-keys main/default-metrics
[:jobs-orphaned :jobs-total-timing])}))
(defn- counter-value [metrics id labels]
(let [collector (mtx/get-collector metrics id)
instance (::mdef/instance collector)
child (.labels ^Counter instance (into-array String labels))]
(.get ^Counter$Child child)))
(defn- histogram-sample-count [metrics labels]
(->> (enumeration-seq
(.metricFamilySamples ^io.prometheus.client.CollectorRegistry
(mtx/get-registry metrics)))
(mapcat (fn [^Collector$MetricFamilySamples family]
(.samples family)))
(filter (fn [^Collector$MetricFamilySamples$Sample sample]
(and (str/ends-with? (.-name sample) "_count")
(= labels (vec (.-labelValues sample))))))
(map (fn [^Collector$MetricFamilySamples$Sample sample]
(long (.-value sample))))
(reduce + 0)))
(defn- mk-job
[{:keys [name queue status scheduled-at modified-at tenant]
:or {name "test-job"
queue "test"
tenant (cf/get :tenant)
status "new"
scheduled-at (ct/now)
modified-at (ct/now)}}]
(let [id (uuid/next)]
(th/db-insert! :job {:id id
:name name
:tenant tenant
:queue queue
:params (db/json {})
:priority 100
:max-retries 3
:retry-num 0
:status status
:scheduled-at scheduled-at
:created-at (ct/now)
:modified-at modified-at})
id))
(defn- get-row
[id]
(-> (th/db-get :job {:id id} :status :modified-at :scheduled-at :error)
(update :error jobs/decode-job-error)
(assoc :events (->> (th/db-exec! ["SELECT kind, payload FROM job_event
WHERE job_id = ? ORDER BY id" id])
(mapv (fn [{:keys [kind payload]}]
{:kind kind
:payload (db/decode-json-pgobject payload)}))))))
(defn- queue-key
[queue-name]
(wrk/queue-key (cf/get :tenant) queue-name))
(defn- test-fixture [next]
(th/database-reset
(fn []
;; clear the redis hand-off list so tests don't see stale payloads
(let [conn (rds/connect (mk-cfg))]
(try
(rds/del conn (queue-key "test"))
(finally
(rds/close conn)))
(next)))))
(t/use-fixtures :each test-fixture)
(defn- drain-queue
[queue-name]
(let [conn (rds/connect (mk-cfg))]
(try
(let [cmd (.-cmd conn)
res (.lrange ^io.lettuce.core.api.sync.RedisCommands
cmd (queue-key queue-name) 0 -1)]
(vec res))
(finally
(rds/del conn (queue-key queue-name))
(rds/close conn)))))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; TESTS
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
(t/deftest dispatcher-claims-due-jobs-and-pushes-json-payload
(let [cfg (mk-cfg)
id (mk-job {})
_ (wdisp/run-batch cfg)
row (get-row id)
key (queue-key "test")]
(t/testing "claimed job is marked scheduled"
(t/is (= "scheduled" (:status row))))
(t/testing "payload is plain JSON [uuid, iso-8601]"
(let [payloads (drain-queue "test")]
(t/is (= 1 (count payloads)))
(let [[job-id scheduled-at :as payload] (json/decode (first payloads))]
(t/is (string? job-id))
(t/is (= (str id) job-id))
(t/is (string? scheduled-at))
(t/testing "round-trip: inst-ms are equal after ct/inst"
(let [expected (:scheduled-at row)
actual (ct/inst scheduled-at)]
(t/is (= (inst-ms expected) (inst-ms actual))))))))))
(t/deftest dispatcher-claims-retry-jobs-too
(let [cfg (mk-cfg)
id (mk-job {:status "retry"})]
(wdisp/run-batch cfg)
(t/is (= "scheduled" (:status (get-row id))))))
(t/deftest dispatcher-does-not-claim-future-or-terminal-jobs
(let [cfg (mk-cfg)
future (mk-job {:scheduled-at (ct/plus (ct/now)
(ct/duration {:minutes 10}))})
running (mk-job {:status "running"})]
(wdisp/run-batch cfg)
(t/is (= "new" (:status (get-row future))))
(t/is (= "running" (:status (get-row running))))))
(t/deftest dispatcher-reschedules-lost-scheduled-jobs
(let [cfg (mk-cfg)
lost-id (mk-job {:status "scheduled"
:scheduled-at (ct/minus (ct/now)
(ct/duration {:minutes 6}))})
fresh-id (mk-job {:status "scheduled"})]
(wdisp/run-batch cfg)
(let [lost (get-row lost-id)
fresh (get-row fresh-id)]
;; the lost job is rescheduled to 'new' and claimed again in the
;; same batch, so it ends as 'scheduled' with a recent scheduled_at
(t/is (= "scheduled" (:status lost)))
(t/is (> (inst-ms (:scheduled-at lost))
(inst-ms (ct/minus (ct/now)
(ct/duration {:minutes 5})))))
(t/testing "the rescheduled job was pushed again to the queue"
(t/is (= 1 (count (drain-queue "test"))))
(t/is (= "scheduled" (:status fresh)))))))
(t/deftest dispatcher-marks-stale-running-jobs-as-aborted-by-lease
(let [cfg (mk-cfg)
stale (mk-job {:status "running"
:modified-at (ct/minus (ct/now)
(ct/plus (cf/get-jobs-lease)
(ct/duration {:minutes 1})))})
fresh (mk-job {:status "running"})]
(wdisp/run-batch cfg)
(let [stale-row (get-row stale)]
(t/is (= "aborted" (:status stale-row)))
(t/testing "the error is the structured one, decoded back to keywords"
(t/is (= jobs/orphan-error (:error stale-row))))
(t/testing "the sweep stores the end event with the aborted outcome"
(t/is (= [{:kind "end" :payload {:outcome "aborted"}}]
(:events stale-row))))
(t/testing "an aborted job is terminal: it is never retried nor reclaimed"
(t/is (zero? (jobs/retry-job cfg
(:id stale-row)
1 (ct/now) jobs/orphan-error :backoff)))))
(t/testing "recent running job is not touched"
(t/is (= "running" (:status (get-row fresh)))))))
(t/deftest dispatcher-orphan-sweep-records-orphan-and-terminal-metrics
(let [metrics (make-orphan-metrics)
cfg (assoc (mk-cfg) ::mtx/metrics metrics)
_ (mk-job {:status "running"
:modified-at (ct/minus (ct/now)
(ct/plus (cf/get-jobs-lease)
(ct/duration {:minutes 1})))})]
(wdisp/run-batch cfg)
(t/is (= 1.0 (counter-value metrics :jobs-orphaned ["test"])))
(t/is (= 1 (histogram-sample-count metrics ["test-job" "test" "aborted"])))))
(t/deftest dispatcher-batch-without-pending-jobs-signals-wait
(let [cfg (mk-cfg)]
(t/is (= ::wdisp/wait (wdisp/run-batch cfg)))))
(t/deftest dispatcher-failed-mark-leaves-no-orphan-payload
;; Mark runs before push: when the mark fails, the push never runs, so
;; a rolled-back batch must leave neither a marked row nor a payload.
(let [cfg (assoc (mk-cfg) ::wdisp/timeout (ct/duration {:millis 10}))
id (mk-job {})
orig @#'wdisp/mark-as-scheduled
calls (atom 0)]
(alter-var-root #'wdisp/mark-as-scheduled
(constantly (fn [& args]
(when (= 1 (swap! calls inc))
(throw (ex-info "boom" {})))
(apply orig args))))
(try
(wdisp/run-batch cfg)
(t/testing "failed mark rolls back with no payload pushed"
(t/is (= "new" (:status (get-row id))))
(t/is (empty? (drain-queue "test"))))
(finally
(alter-var-root #'wdisp/mark-as-scheduled (constantly orig))))
(t/testing "next batch delivers exactly once"
(wdisp/run-batch cfg)
(t/is (= "scheduled" (:status (get-row id))))
(t/is (= 1 (count (drain-queue "test")))))))
(t/deftest dispatcher-records-one-redis-failure
(let [metrics (make-metrics)
cfg (assoc (mk-cfg) ::mtx/metrics metrics
::wdisp/timeout (ct/duration {:millis 10}))
_ (mk-job {})]
(with-redefs [rds/rpush (fn [& _] (throw (ex-info "redis down" {})))]
(wdisp/run-batch cfg))
(t/is (= 1 (histogram-sample-count metrics ["execution" "failed"])))
(t/is (= 0 (histogram-sample-count metrics ["dispatch" "failed"])))))
(t/deftest dispatcher-push-failure-rolls-back-without-throwing
(let [cfg (assoc (mk-cfg) ::wdisp/timeout (ct/duration {:millis 10}))
id (mk-job {})
orig @#'rds/rpush]
(alter-var-root #'rds/rpush
(constantly (fn [& _] (throw (ex-info "redis down" {})))))
(try
;; run-batch catches the failure (sleep path) instead of throwing;
;; a throw would error this test by itself
(wdisp/run-batch cfg)
(t/testing "failed push rolls back the batch"
(t/is (= "new" (:status (get-row id))))
(t/is (empty? (drain-queue "test"))))
(finally
(alter-var-root #'rds/rpush (constantly orig))))))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; TENANT ISOLATION
;; A batch is this tenant's: it must not read, claim or kill the rows of
;; another instance sharing the same database.
(t/deftest dispatcher-does-not-claim-another-tenants-jobs
(let [cfg (mk-cfg)
foreign (mk-job {:tenant "other-tenant"})]
(wdisp/run-batch cfg)
(t/testing "the other tenant's job is not claimed"
(t/is (= "new" (:status (get-row foreign)))))
(t/testing "and nothing was pushed to this tenant's Redis list"
(t/is (empty? (drain-queue "test"))))))
(t/deftest dispatcher-does-not-reschedule-another-tenants-lost-jobs
(let [cfg (mk-cfg)
foreign (mk-job {:status "scheduled"
:tenant "other-tenant"
:scheduled-at (ct/minus (ct/now)
(ct/duration {:minutes 6}))})]
(wdisp/run-batch cfg)
(t/testing "the other tenant's scheduled job keeps its status and its date"
(let [row (get-row foreign)]
(t/is (= "scheduled" (:status row)))
(t/is (< (inst-ms (:scheduled-at row))
(inst-ms (ct/minus (ct/now)
(ct/duration {:minutes 5})))))))))
(t/deftest dispatcher-does-not-orphan-another-tenants-running-jobs
(let [cfg (mk-cfg)
;; older than any lease: this tenant's sweep would kill it
foreign (mk-job {:status "running"
:tenant "other-tenant"
:modified-at (ct/minus (ct/now)
(ct/plus (cf/get-jobs-lease)
(ct/duration {:minutes 1})))})]
(wdisp/run-batch cfg)
(t/testing "the other tenant's running job is still running"
(let [row (get-row foreign)]
(t/is (= "running" (:status row)))
(t/is (nil? (:error row)))
(t/is (= [] (:events row)))))))
Reference in New Issue View Git Blame Copy Permalink