diff --git a/backend/src/app/worker/cron.clj b/backend/src/app/worker/cron.clj index 4e797006dd..0105d9d57e 100644 --- a/backend/src/app/worker/cron.clj +++ b/backend/src/app/worker/cron.clj @@ -12,8 +12,8 @@ detection, retries and telemetry as for every other job. To keep the no-overlap semantics (one entry is not re-fired while the previous instance lives), the scheduler only submits when there is no active - instance (`new`/`scheduled`/`running`/`retry` with the same `name` - and `label` of the entry). The `FOR UPDATE SKIP LOCKED` claim on the + instance (`new`/`scheduled`/`running`/`retry` with the same `name`, + `queue` and `label` of the entry). The `FOR UPDATE SKIP LOCKED` claim on the scheduled_task row already serializes the tick between nodes." (:require [app.common.data :as d] @@ -41,18 +41,6 @@ on conflict (id) do nothing") -;; The tenant filter is what makes the no-overlap check per instance: the -;; job this count decides about is submitted to this tenant's queue, so -;; counting other tenants' active jobs would let one instance skip a tick -;; for work that is none of its business. -(def ^:private sql:count-active-jobs - "SELECT count(*) AS n - FROM job - WHERE name = ? - AND label = ? - AND tenant = ? - AND status IN ('new','scheduled','running','retry')") - (defn- synchronize-cron-entries [{:keys [::db/conn ::entries]}] (doseq [{:keys [id cron]} entries] @@ -70,14 +58,16 @@ (defn submit-cron-job "Submit the system job for the entry to the `:cron` queue. Uses the entry id as the job label (stable per entry) and returns the created - job-id. No-overlap pre-check lives in the caller." + job-id, or nil when an active instance (`new`/`scheduled`/`running`/ + `retry` with the same name, queue and label) already exists." [cfg {:keys [job props id]}] (jobs/submit cfg - {::jobs/name job - ::jobs/params (or props {}) ;; entries don't carry props; default to empty map - ::jobs/queue :cron - ::jobs/label (name id)})) + {::jobs/name job + ::jobs/params (or props {}) ;; entries don't carry props; default to empty map + ::jobs/queue :cron + ::jobs/label (name id) + ::jobs/skip-if-active true})) (declare ^:private schedule-cron-task) @@ -100,29 +90,28 @@ {:id id} {::db/return-keys false}) - ;; The count query is not locked, so a job could - ;; transition from running to completed between the - ;; count and the submit. This is acceptable: the - ;; next tick will submit if needed. The FOR UPDATE - ;; SKIP LOCKED on the scheduled_task row prevents - ;; race conditions between nodes. - (let [tenant (cf/get :tenant) - row (db/exec-one! conn [sql:count-active-jobs (d/name job) (str id) tenant]) - active (or (get row :n) 0)] - (if (pos? active) - (do - (db/after-commit #(metrics/record-cron cfg :skipped :active)) - (l/dbg :hint "skip" - :reason "scheduling, active instance exists" - :id id :job (d/name job))) - (let [job-id (submit-cron-job cfg entry) - elapsed (ct/format-duration (tpoint))] - (db/after-commit #(metrics/record-cron cfg :submitted :none)) - (l/dbg :hint "submit" - :id id - :job (d/name job) - :job-id (str job-id) - :elapsed elapsed))))))) + ;; The no-overlap check lives in the submit itself + ;; (`::jobs/skip-if-active`): the check and the + ;; insert share this transaction, so a job could + ;; still transition from running to completed + ;; between the check and the insert. This is + ;; acceptable: the next tick will submit if + ;; needed. The FOR UPDATE SKIP LOCKED on the + ;; scheduled_task row prevents race conditions + ;; between nodes. + (if-let [job-id (submit-cron-job cfg entry)] + (let [elapsed (ct/format-duration (tpoint))] + (db/after-commit #(metrics/record-cron cfg :submitted :none)) + (l/dbg :hint "submit" + :id id + :job (d/name job) + :job-id (str job-id) + :elapsed elapsed)) + (do + (db/after-commit #(metrics/record-cron cfg :skipped :active)) + (l/dbg :hint "skip" + :reason "scheduling, active instance exists" + :id id :job (d/name job))))))) (catch InterruptedException _ (metrics/record-cron cfg :error :interrupted) diff --git a/backend/test/backend_tests/worker_cron_test.clj b/backend/test/backend_tests/worker_cron_test.clj index d8ab47209a..e0797ab4b0 100644 --- a/backend/test/backend_tests/worker_cron_test.clj +++ b/backend/test/backend_tests/worker_cron_test.clj @@ -199,3 +199,51 @@ (t/testing "and the other tenant's row is untouched" (t/is (= "running" (:status (th/db-get :job {:name "session-gc" :tenant "other-tenant"}))))))) + +(t/deftest cron-tick-is-not-blocked-by-active-job-on-another-queue + (let [cfg (make-cfg) + entry {:id "session-gc-other-queue" + :job :session-gc + :cron (ucron/cron "0 0 0 * * ?")}] + + (th/db-insert! :scheduled-task {:id "session-gc-other-queue" + :cron-expr "0 0 * * *"}) + ;; A live job with the same name, label and tenant, but sitting on + ;; another queue. The cron no-overlap check is qualified by queue, + ;; so it is none of this tick's business. + (th/db-insert! :job {:id (uuid/next) + :name "session-gc" + :tenant (cf/get :tenant) + :queue "default" + :label "session-gc-other-queue" + :status "running" + :params (db/json {}) + :scheduled-at (ct/now) + :created-at (ct/now) + :modified-at (ct/now)}) + + (.join ^Thread (#'cron/execute-cron-task cfg entry)) + + (t/testing "the tick still submits its own job on the cron queue" + (t/is (= 1 (count-jobs cfg :name "session-gc" + :queue "cron" + :label "session-gc-other-queue")))) + (t/testing "and the row on the other queue is untouched" + (t/is (= "running" (:status (th/db-get :job {:name "session-gc" + :queue "default" + :label "session-gc-other-queue"}))))))) + +(t/deftest cron-submit-skips-when-active-instance-on-cron-queue + (let [cfg (make-cfg) + entry (first (get-cron-entries))] + + (let [job-id (cron/submit-cron-job cfg entry)] + (t/is (some? job-id)) + + (t/testing "a second submit while the first is live returns nil" + (t/is (nil? (cron/submit-cron-job cfg entry)))) + + (t/testing "no second row was inserted" + (t/is (= 1 (count-jobs cfg :name "session-gc" + :queue "cron" + :label "session-gc-no-props")))))))