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)]