From 5eb33add7ee7c15c4cbeab67b0917773ccbdcb63 Mon Sep 17 00:00:00 2001 From: Andrey Antukh Date: Sun, 11 Oct 2026 18:48:02 +0200 Subject: [PATCH] :sparkles: Add skip-if-active option to jobs submit (#12266) * :sparkles: 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 * :sparkles: 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 * :recycle: 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 --- .serena/memories/backend/subtleties.md | 3 +- backend/src/app/jobs.clj | 77 ++++++++++++++++--- backend/src/app/jobs/metrics.clj | 16 ++-- backend/src/app/main.clj | 26 +++---- backend/src/app/worker/dispatcher.clj | 1 - .../test/backend_tests/jobs_metrics_test.clj | 45 ++++++++--- .../test/backend_tests/jobs_request_test.clj | 31 +++++--- backend/test/backend_tests/jobs_test.clj | 63 +++++++++++++++ .../backend_tests/worker_dispatcher_test.clj | 6 +- 9 files changed, 204 insertions(+), 64 deletions(-) diff --git a/.serena/memories/backend/subtleties.md b/.serena/memories/backend/subtleties.md index 3e1b73a482..4d066393b2 100644 --- a/.serena/memories/backend/subtleties.md +++ b/.serena/memories/backend/subtleties.md @@ -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 `:` 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::` 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. diff --git a/backend/src/app/jobs.clj b/backend/src/app/jobs.clj index be36f25549..19df02cc9b 100644 --- a/backend/src/app/jobs.clj +++ b/backend/src/app/jobs.clj @@ -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 diff --git a/backend/src/app/jobs/metrics.clj b/backend/src/app/jobs/metrics.clj index a1dd8a2343..5ba7df2864 100644 --- a/backend/src/app/jobs/metrics.clj +++ b/backend/src/app/jobs/metrics.clj @@ -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))))) diff --git a/backend/src/app/main.clj b/backend/src/app/main.clj index 87d0b31f29..619fa9a150 100644 --- a/backend/src/app/main.clj +++ b/backend/src/app/main.clj @@ -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]}}) diff --git a/backend/src/app/worker/dispatcher.clj b/backend/src/app/worker/dispatcher.clj index 3ade655db3..b084966714 100644 --- a/backend/src/app/worker/dispatcher.clj +++ b/backend/src/app/worker/dispatcher.clj @@ -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)))) diff --git a/backend/test/backend_tests/jobs_metrics_test.clj b/backend/test/backend_tests/jobs_metrics_test.clj index 45f4b59725..7b2523c320 100644 --- a/backend/test/backend_tests/jobs_metrics_test.clj +++ b/backend/test/backend_tests/jobs_metrics_test.clj @@ -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) diff --git a/backend/test/backend_tests/jobs_request_test.clj b/backend/test/backend_tests/jobs_request_test.clj index bb45c500b3..976102def1 100644 --- a/backend/test/backend_tests/jobs_request_test.clj +++ b/backend/test/backend_tests/jobs_request_test.clj @@ -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) diff --git a/backend/test/backend_tests/jobs_test.clj b/backend/test/backend_tests/jobs_test.clj index 20675fed2b..5dee48bbb3 100644 --- a/backend/test/backend_tests/jobs_test.clj +++ b/backend/test/backend_tests/jobs_test.clj @@ -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) diff --git a/backend/test/backend_tests/worker_dispatcher_test.clj b/backend/test/backend_tests/worker_dispatcher_test.clj index 02064f2774..c498214beb 100644 --- a/backend/test/backend_tests/worker_dispatcher_test.clj +++ b/backend/test/backend_tests/worker_dispatcher_test.clj @@ -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)]