github/penpot
1
0
Fork 0
mirror of https://github.com/penpot/penpot.git synced 2026-10-28 21:40:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

✨ Add skip-if-active option to jobs submit (#12266)

Browse Source 
* ✨ 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
This commit is contained in:
Andrey Antukh 2026-10-11 18:48:02 +02:00 committed by GitHub
parent 10955f17e2
commit 5eb33add7e
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
9 changed files with 204 additions and 64 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
.serena/memories/backend/subtleties.md
View File

@ -24,7 +24,8 @@
## Workers and cron
- `job` carries `tenant` (the instance the job belongs to) and a bare `queue`. Several instances share one database and the tenant is what keeps an instance from claiming another environment's jobs, so every query that acts on one instance's work filters on `tenant`: dispatch claim, orphan sweep, lost-job reschedule, cron no-overlap count, submit dedupe. The `job__dispatcher__idx` leads with `tenant` (equality) then `(priority DESC, scheduled_at)`; `queue` does not lead because the dispatcher claims every queue of its tenant. The `<tenant>:<queue>` prefix survives only in the Redis key, composed once by `app.worker/queue-key`.
- Submit dedupe only removes not-yet-due `new` jobs of the same tenant with the same name/queue/label; it does not dedupe due, scheduled, retry, running, or completed work.
- Submit dedupe only removes not-yet-due `new` jobs of the same tenant with the same name/queue/label; it does not dedupe due, scheduled, retry, running, or completed work. The `::jobs/skip-if-active` submit option goes further: it skips the insert (returning `nil`, recording `penpot_jobs_skipped_total`) when a live job (`new`/`scheduled`/`running`/`retry`) already exists for the same tenant/name/queue/label, requires an explicit non-blank label, and carries the same best-effort cross-backend race caveat as dedupe.
- Jobs metrics rule (one event, one family): counters exist only for events without latency; when an event has a histogram, its `_count` is the counter (no twin counter). `dispatched` (pushes) vs `queue-wait` (claims) are deliberately different events — do not merge them.
- The dispatcher selects `new`/`retry` jobs with `FOR UPDATE SKIP LOCKED`, marks them `scheduled`, and publishes the JSON payload `[id scheduled-at]` to the `penpot.worker.queue:<tenant>:<queue>` Redis list. The runner skips Redis messages whose scheduled timestamp no longer matches DB state.
- Lost `scheduled` jobs are rescheduled after 5 minutes; `running` jobs untouched longer than `:jobs-lease` (default 30 min) are marked `aborted` with `app.jobs/orphan-error` (`type :internal`, `code :orphan`). `aborted` is a system-side terminal state like `cancelled` (never retried, swept by retention like other terminals, reported with an error log); long jobs must `heartbeat` per chunk so a healthy backlog is never mistaken for a dead process. The `:progress` report is optional and only feeds the durable log. A beat whose write finds the job no longer active raises an `:interrupt` (`code :job-interrupted`) instead of counting zero: it is the internal stop-the-work signal, not an error, so neither the listener nor the runner logs it as one.
- An orphan gets its `end` event (`outcome "aborted"`) from the sweep itself: one bulk `INSERT ... SELECT` in the same transaction as the bulk `UPDATE`, never one insert per orphan. No msgbus notification goes out for it; orphans alert through the error log and the orphaned counter.

77
backend/src/app/jobs.clj
View File

@ -245,6 +245,19 @@
AND status = 'new'
AND scheduled_at > ?")
;; The cron no-overlap check and the skip-if-active submit option share
;; this shape: both look for jobs of one tenant with the same name, queue
;; and label that have not reached a terminal state yet. The
;; `job__name_label__idx` partial index serves both queries.
(def ^:private sql:count-active-jobs
"SELECT count(*) AS n
FROM job
WHERE name = ?
AND tenant = ?
AND queue = ?
AND label = ?
AND status IN ('new', 'scheduled', 'running', 'retry')")
(def ^:private schema:options
[:map {:title "submit-options"}
[::name schema:job-name]
@ -255,6 +268,10 @@
[::priority {:optional true} ::sm/int]
[::max-retries {:optional true} ::sm/int]
[::dedupe {:optional true} ::sm/boolean]
;; Skip the insert when a live job already covers the same
;; (tenant, name, queue, label). Requires an explicit label: the
;; default empty one would block every flagged submission at once.
[::skip-if-active {:optional true} ::sm/boolean]
;; Owner of a user-facing job. Its events are published on the msgbus
;; topic of this profile; internal jobs omit it.
[::profile-id {:optional true} ::sm/uuid]
@ -289,17 +306,36 @@
`::expires-at`) are validated and stored as columns; nothing is inferred
from `params`.
With `::skip-if-active`, the insert is skipped (and `nil` returned)
when a live job (`new`, `scheduled`, `running` or `retry`) already
exists for the same tenant, name, queue and label. The check and the
insert share one transaction. A skip deletes nothing and inserts
nothing; it records a skipped metric instead of a submitted one, so
dashboards can tell withheld work apart from new work. The option
needs an explicit non-blank `::label`, otherwise it raises.
NOTE: the dedupe DELETE and the INSERT run atomically: joined to
the caller's transaction when the cfg provides `::db/conn`, wrapped
in their own transaction otherwise. Concurrent cross-backend
submissions can, in rare race conditions, produce duplicated 'new'
rows (accepted risk, see prod-infra documentation)."
rows (accepted risk, see prod-infra documentation). The same caveat
covers the skip-if-active check: a race between backends can slip
two inserts past the count."
[cfg {:keys [::params ::name ::delay ::queue ::priority ::max-retries
::dedupe ::label ::profile-id ::resource-id ::expires-at]
::dedupe ::skip-if-active ::label
::profile-id ::resource-id ::expires-at]
:or {delay 0 queue :default priority 100 max-retries 3 label ""}
:as options}]
(assert (mtx/instance cfg) "missing ::mtx/metrics on the jobs cfg")
;; Before the schema check so the failure names the option, not the
;; shape: the schema would reject a blank label with a generic
;; "check error" either way.
(when (and skip-if-active (or (not (string? label)) (str/blank? label)))
(ex/raise :type :validation
:code :skip-if-active-requires-label
:hint "skip-if-active requires an explicit non-blank label"
:name (d/name name)))
(check-options options)
(let [job-def (get-job-def (get-defs cfg) name)
params (validate-params job-def params)
@ -327,6 +363,14 @@
;; between backends could create duplicates, but this is
;; acceptable: cross-backend races are rare, jobs are idempotent,
;; and dedupe is best-effort.
;; Skip-if-active is best-effort like dedupe: the count and the
;; insert share this transaction, but a race between backends
;; can still slip two inserts past the count.
active? (fn [conn]
(when skip-if-active
(pos? (or (:n (db/exec-one! conn [sql:count-active-jobs
job-name tenant queue label]))
0))))
insert! (fn [conn]
(let [deleted (when dedupe
;; the DELETE below is intentional:
@ -353,13 +397,23 @@
;; caller's transaction when the cfg provides a connection,
;; wrapped in their own otherwise (a failed INSERT can never
;; orphan a committed DELETE, even on an autocommit caller conn).
;; A skip returns nil: no DELETE, no INSERT, no submitted metric.
(db/tx-run! cfg
(fn [{:keys [::db/conn]}]
(let [result (insert! conn)]
(db/after-commit
#(metrics/record-submitted cfg job-name queue))
result)))
id))
(if (active? conn)
(do (db/after-commit
#(metrics/record-skipped cfg job-name queue))
(l/trc :hint "submit skipped, active job exists"
:name job-name
:tenant tenant
:queue queue
:label label
:skipped true))
(let [result (insert! conn)]
(db/after-commit
#(metrics/record-submitted cfg job-name queue))
result
id))))))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; JOB API
@ -845,13 +899,14 @@
nil)))
(defn- record-terminal
"Record the terminal outcome and the total time. Runs after the commit
"Record the total time of a terminal outcome. Runs after the commit
that made the job terminal, so a rolled back transition is never
counted. Takes the cfg and resolves the metrics instance from it, so
the callback is not tied to an instance read at registration time."
counted. The histogram `_count` is the terminal total per outcome:
no separate counter. Takes the cfg and resolves the metrics
instance from it, so the callback is not tied to an instance read
at registration time."
[cfg job outcome]
(when job
(metrics/record-outcome cfg (:name job) (:queue job) outcome)
(when (ct/inst? (:created-at job))
(metrics/record-total
cfg

16
backend/src/app/jobs/metrics.clj
View File

@ -65,19 +65,18 @@
[(name-label name) (queue-label queue)]
1))
(defn record-skipped
[cfg name queue]
(record-count (mtx/instance cfg) :jobs-skipped
[(name-label name) (queue-label queue)]
1))
(defn record-dispatched
[cfg name queue amount]
(record-count (mtx/instance cfg) :jobs-dispatched
[(name-label name) (queue-label queue)]
amount))
(defn record-outcome
[cfg name queue outcome]
(record-count (mtx/instance cfg) :jobs-completed
[(name-label name) (queue-label queue)
(outcome-label outcome)]
1))
(defn record-retry
[cfg name queue reason]
(record-count (mtx/instance cfg) :jobs-retries
@ -172,7 +171,8 @@
[cfg outcome millis]
(let [outcome (str/lower (or (some-> outcome d/name) "error"))
metrics (mtx/instance cfg)]
(record-count metrics :jobs-requests-total [outcome] 1)
;; No separate counter: the timing `_count` is the request total
;; per outcome.
(record-value metrics :jobs-request-timing [outcome]
(max 0 (long millis)))))

26
backend/src/app/main.clj
View File

@ -240,16 +240,16 @@
::mdef/labels ["name" "queue"]
::mdef/type :counter}
:jobs-dispatched
{::mdef/name "penpot_jobs_dispatched_total"
::mdef/help "Total number of jobs pushed to a worker queue."
:jobs-skipped
{::mdef/name "penpot_jobs_skipped_total"
::mdef/help "Total number of durable job submissions skipped by skip-if-active."
::mdef/labels ["name" "queue"]
::mdef/type :counter}
:jobs-completed
{::mdef/name "penpot_jobs_completed_total"
::mdef/help "Total number of durable jobs reaching a terminal state."
::mdef/labels ["name" "queue" "outcome"]
:jobs-dispatched
{::mdef/name "penpot_jobs_dispatched_total"
::mdef/help "Total number of jobs pushed to a worker queue. Counts pushes, not claims: compare with the _count of penpot_jobs_queue_wait_ms."
::mdef/labels ["name" "queue"]
::mdef/type :counter}
:jobs-retries
@ -277,7 +277,7 @@
:jobs-queue-wait-timing
{::mdef/name "penpot_jobs_queue_wait_ms"
::mdef/help "Time from job scheduling to worker claim (milliseconds)."
::mdef/help "Time from job scheduling to worker claim (milliseconds). Its _count counts claims, not pushes: compare with penpot_jobs_dispatched_total."
::mdef/labels ["name" "queue"]
::mdef/type :histogram
::mdef/buckets [1 10 50 100 500 1000 5000 10000 30000 60000 300000 600000]}
@ -291,7 +291,7 @@
:jobs-total-timing
{::mdef/name "penpot_jobs_total_ms"
::mdef/help "Time from job creation to a terminal state (milliseconds)."
::mdef/help "Time from job creation to a terminal state (milliseconds). Its _count is the terminal total per outcome."
::mdef/labels ["name" "queue" "outcome"]
::mdef/type :histogram
::mdef/buckets [10 50 100 500 1000 5000 10000 30000 60000 300000 600000 3600000]}
@ -340,15 +340,9 @@
::mdef/labels ["outcome" "reason"]
::mdef/type :counter}
:jobs-requests-total
{::mdef/name "penpot_jobs_requests_total"
::mdef/help "Total number of ephemeral job request events."
::mdef/labels ["outcome"]
::mdef/type :counter}
:jobs-request-timing
{::mdef/name "penpot_jobs_request_ms"
::mdef/help "Time spent waiting for an ephemeral job reply (milliseconds)."
::mdef/help "Time spent waiting for an ephemeral job reply (milliseconds). Its _count is the request total per outcome."
::mdef/labels ["outcome"]
::mdef/type :histogram
::mdef/buckets [1 10 50 100 500 1000 5000 10000 30000 60000 300000]}})

1
backend/src/app/worker/dispatcher.clj
View File

@ -135,7 +135,6 @@ RETURNING job.id, job.queue, job.name, job.created_at")
(fn []
(doseq [{:keys [id queue name created-at]} rows]
(metrics/record-orphan cfg queue)
(metrics/record-outcome cfg name queue :aborted)
(when (ct/inst? created-at)
(metrics/record-total cfg name queue :aborted
(- (inst-ms timestamp) (inst-ms created-at))))

45
backend/test/backend_tests/jobs_metrics_test.clj
View File

@ -33,8 +33,8 @@
(def ^:private metric-ids
[:jobs-submitted
:jobs-skipped
:jobs-dispatched
:jobs-completed
:jobs-retries
:jobs-orphaned
:jobs-rescheduled
@ -48,7 +48,6 @@
:jobs-gc-rows
:jobs-gc-timing
:jobs-cron-total
:jobs-requests-total
:jobs-request-timing
:jobs-events-total])
@ -103,7 +102,7 @@
(t/is (some? definition) id)
(t/is (re-find #"^penpot_jobs_" (::mdef/name definition)) id)))
(t/is (= ["name" "queue" "outcome"]
(::mdef/labels (:jobs-completed main/default-metrics))))
(::mdef/labels (:jobs-total-timing main/default-metrics))))
(t/is (= ["status"]
(::mdef/labels (:jobs-backlog main/default-metrics)))))
@ -140,14 +139,37 @@
(jobs/claim cfg job-id (:scheduled-at (jobs/get-job cfg job-id)))
(jobs/complete cfg :job-id job-id :result {:ok true})
(t/is (= 1.0 (counter-value metrics :jobs-submitted ["echo" "default"])))
(t/is (= 1.0 (counter-value metrics :jobs-completed ["echo" "default" "completed"])))))
(t/is (= 1.0 (histogram-count metrics :jobs-total-timing ["echo" "default" "completed"])))))
(t/deftest skip-if-active-records-skipped-metric
(let [metrics (make-metrics)
defs {:echo {::jobs/name :echo
::jobs/schema [:map]
::jobs/handler (fn [_context params] params)
::jobs/decoder identity
::jobs/validator (constantly true)}}
cfg {::jobs/defs defs
::db/pool th/*pool*
::mtx/metrics metrics}
_ (jobs/submit cfg {::jobs/name :echo
::jobs/params {}
::jobs/label "skip-metric"})
skipped (jobs/submit cfg {::jobs/name :echo
::jobs/params {}
::jobs/label "skip-metric"
::jobs/skip-if-active true})]
(t/is (nil? skipped))
(t/is (= 1.0 (counter-value metrics :jobs-skipped ["echo" "default"])))
(t/is (= 1.0 (counter-value metrics :jobs-submitted ["echo" "default"]))
"the withheld submit records no submission")))
(t/deftest lifecycle-helpers-use-labels-as-is
(let [metrics (make-metrics)
cfg (metrics-cfg metrics)]
(jobs-metrics/record-submitted cfg "delete-object" "default")
(jobs-metrics/record-skipped cfg "delete-object" "default")
(jobs-metrics/record-dispatched cfg "delete-object" "default" 2)
(jobs-metrics/record-outcome cfg "delete-object" "default" :completed)
(jobs-metrics/record-total cfg "delete-object" "default" :completed 42)
(jobs-metrics/record-retry cfg "delete-object" "default" :backoff)
(jobs-metrics/record-orphan cfg "webhooks")
(jobs-metrics/record-rescheduled cfg "custom")
@ -157,14 +179,15 @@
(jobs-metrics/record-legacy-execution cfg "delete-object" 42)
(t/is (= 1.0 (counter-value metrics :jobs-submitted ["delete-object" "default"])))
(t/is (= 1.0 (counter-value metrics :jobs-skipped ["delete-object" "default"])))
(t/is (= 2.0 (counter-value metrics :jobs-dispatched ["delete-object" "default"])))
(t/is (= 1.0 (counter-value metrics :jobs-completed ["delete-object" "default" "completed"])))
(t/is (= 1.0 (histogram-count metrics :jobs-total-timing ["delete-object" "default" "completed"])))
(t/is (= 1.0 (counter-value metrics :jobs-retries ["delete-object" "default" "backoff"])))
(t/is (= 1.0 (counter-value metrics :jobs-orphaned ["webhooks"])))
(t/is (= 1.0 (counter-value metrics :jobs-rescheduled ["custom"])))
(t/is (= 3.0 (counter-value metrics :jobs-gc-rows ["expired" "deleted"])))
(t/is (= 1.0 (counter-value metrics :jobs-cron-total ["submitted" "none"])))
(t/is (= 1.0 (counter-value metrics :jobs-requests-total ["replied"])))
(t/is (= 1.0 (histogram-count metrics :jobs-request-timing ["replied"])))
(t/testing "the legacy histogram keeps the raw job name, with no queue"
(t/is (= 1.0 (histogram-count metrics :tasks-timing ["delete-object"]))))
@ -181,15 +204,15 @@
(t/is (= "other" (jobs-metrics/queue-label nil))))
(let [metrics (make-metrics)
cfg (metrics-cfg metrics)]
(jobs-metrics/record-outcome cfg "echo" "unexpected" :unexpected)
(t/is (= 1.0 (counter-value metrics :jobs-completed ["echo" "unexpected" "unexpected"])))
(jobs-metrics/record-total cfg "echo" "unexpected" :unexpected 42)
(t/is (= 1.0 (histogram-count metrics :jobs-total-timing ["echo" "unexpected" "unexpected"])))
(t/testing "a new value is recorded, never folded or dropped"
(jobs-metrics/record-gc-rows cfg :unknown :deleted 1)
(jobs-metrics/record-cron cfg :brand-new :whatever)
(jobs-metrics/record-request cfg :brand-new 5)
(t/is (= 1.0 (counter-value metrics :jobs-gc-rows ["unknown" "deleted"])))
(t/is (= 1.0 (counter-value metrics :jobs-cron-total ["brand-new" "whatever"])))
(t/is (= 1.0 (counter-value metrics :jobs-requests-total ["brand-new"]))))))
(t/is (= 1.0 (histogram-count metrics :jobs-request-timing ["brand-new"]))))))
(t/deftest sampler-is-wired-and-does-not-start-on-read-only
(t/is (contains? main/worker-config :app.jobs.metrics/sampler))
@ -279,7 +302,7 @@
(t/is (= 2 (:cnt (th/db-exec-one! ["SELECT count(*) AS cnt FROM job_event"])))))
(t/testing "the terminal counter followed its own commit, not the caller one"
(t/is (= 1.0 (counter-value metrics :jobs-completed ["echo" "default" "completed"]))))))
(t/is (= 1.0 (histogram-count metrics :jobs-total-timing ["echo" "default" "completed"]))))))
(t/deftest progress-counter-follows-the-report-that-outlives-the-caller
(let [metrics (make-metrics)

31
backend/test/backend_tests/jobs_request_test.clj
View File

@ -23,8 +23,6 @@
[integrant.core :as ig])
(:import
io.lettuce.core.api.sync.RedisCommands
io.prometheus.client.Counter
io.prometheus.client.Counter$Child
java.lang.AutoCloseable))
(t/use-fixtures :once th/state-init)
@ -36,8 +34,7 @@
(defn- make-metrics []
(ig/init-key :app.metrics/metrics
{:default (select-keys main/default-metrics
[:jobs-requests-total
:jobs-request-timing])}))
[:jobs-request-timing])}))
(defn- make-pool
([] (make-pool (get th/*system* :app.metrics/metrics)))
@ -131,11 +128,21 @@
;; TESTS
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
(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-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)
@ -154,14 +161,14 @@
(jobs/request cfg {::jobs/queue :media
::jobs/cmd :process
::jobs/params {}})))
(t/is (= 1.0 (counter-value metrics :jobs-requests-total ["replied"])))
(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 (counter-value metrics :jobs-requests-total ["timeout"])))
(t/is (= 0.0 (counter-value metrics :jobs-requests-total ["sent"])))))
(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)

63
backend/test/backend_tests/jobs_test.clj
View File

@ -507,6 +507,69 @@
AND tenant = ?"
label (cf/get :tenant)])))))))
(t/deftest submit-skip-if-active-skips-live-jobs
(doseq [status ["new" "scheduled" "running" "retry"]]
(t/testing (str "live status " status)
(let [cfg (make-cfg (get-job-defs))
params (make-params)
label (str "skip-active-" status)
id1 (jobs/submit cfg {::jobs/name :echo
::jobs/params params
::jobs/label label})
_ (th/db-update! :job {:status status} {:id id1})
id2 (jobs/submit cfg {::jobs/name :echo
::jobs/params params
::jobs/label label
::jobs/skip-if-active true})]
(t/is (nil? id2))
(t/testing "no row deleted, none inserted"
(t/is (some? (jobs/get-job cfg id1)))
(t/is (= 1 (:cnt (th/db-exec-one! ["SELECT count(*) AS cnt FROM job
WHERE label = ?" label])))))))))
(t/deftest submit-skip-if-active-inserts-past-terminal-jobs
(doseq [status ["completed" "failed" "cancelled" "aborted"]]
(t/testing (str "terminal status " status)
(let [cfg (make-cfg (get-job-defs))
params (make-params)
label (str "skip-terminal-" status)
id1 (jobs/submit cfg {::jobs/name :echo
::jobs/params params
::jobs/label label})
_ (th/db-update! :job {:status status} {:id id1})
id2 (jobs/submit cfg {::jobs/name :echo
::jobs/params params
::jobs/label label
::jobs/skip-if-active true})]
(t/is (uuid? id2))
(t/is (= 2 (:cnt (th/db-exec-one! ["SELECT count(*) AS cnt FROM job
WHERE label = ?" label]))))))))
(t/deftest submit-skip-if-active-requires-an-explicit-label
(let [cfg (make-cfg (get-job-defs))
params (make-params)]
(t/is (thrown-with-msg? Exception #"skip-if-active requires"
(jobs/submit cfg {::jobs/name :echo
::jobs/params params
::jobs/skip-if-active true})))
(t/is (thrown-with-msg? Exception #"skip-if-active requires"
(jobs/submit cfg {::jobs/name :echo
::jobs/params params
::jobs/label " "
::jobs/skip-if-active true})))))
(t/deftest submit-skip-if-active-inserts-when-nothing-is-live
(let [cfg (make-cfg (get-job-defs))
params (make-params)
label "skip-idle-label"
id1 (jobs/submit cfg {::jobs/name :echo
::jobs/params params
::jobs/label label
::jobs/skip-if-active true})]
(t/is (uuid? id1))
(t/is (= 1 (:cnt (th/db-exec-one! ["SELECT count(*) AS cnt FROM job
WHERE label = ?" label]))))))
(t/deftest submit-dedupe-rolls-back-on-insert-failure
(let [cfg (make-cfg (get-job-defs))
params (make-params)

6
backend/test/backend_tests/worker_dispatcher_test.clj
View File

@ -52,8 +52,7 @@
(defn- make-orphan-metrics []
(ig/init-key :app.metrics/metrics
{:default (select-keys main/default-metrics
[:jobs-orphaned :jobs-completed
:jobs-total-timing])}))
[:jobs-orphaned :jobs-total-timing])}))
(defn- counter-value [metrics id labels]
(let [collector (mtx/get-collector metrics id)
@ -228,8 +227,7 @@
(ct/duration {:minutes 1})))})]
(wdisp/run-batch cfg)
(t/is (= 1.0 (counter-value metrics :jobs-orphaned ["test"])))
(t/is (= 1.0 (counter-value metrics :jobs-completed
["test-job" "test" "aborted"])))))
(t/is (= 1 (histogram-sample-count metrics ["test-job" "test" "aborted"])))))
(t/deftest dispatcher-batch-without-pending-jobs-signals-wait
(let [cfg (mk-cfg)]