github/penpot
1
0
Fork 0
mirror of https://github.com/penpot/penpot.git synced 2026-11-09 11:30:54 +00:00
Code Issues Packages Projects Releases Wiki Activity
penpot/backend/test/backend_tests/jobs_test.clj
Andrey Antukh 57fa280209
🐛 Validate and decode a job result with its declared schema (#12350) 
* 🐛 Decode job results with the schema of the job-def

A job result is stored as plain JSON, so a uuid comes back as the text
it is on disk and a uuid used as a map key comes back as a keyword. The
import resolution is a map keyed by file id, and the wizard lost that
identity: choosing a library in the link step created no link at all,
with no error and with the modal closing as if it had worked.

A job-def can now declare a result schema and its precompiled decoder,
the pair the params already have, and the substrate applies them when the
row is read. A def that declares neither answers with the JSON the row
holds, as before; a row whose job-def is gone is read untouched instead
of raising.

Relates to #12339

AI-assisted-by: deepseek-v4.1-flash
Signed-off-by: Andrey Antukh <niwi@niwi.nz>

* 🐛 Type the import result so its resolution keeps uuid keys

The resolution of a version 3 import is a map keyed by the new file id,
and the result column is plain JSON: without a schema of its own every
key came back as a keyword while the :id of the entry came back as text,
so the dashboard wizard could not match the file it was linking and
created no library link at all.

Declare the shape of the resolution (files, done and pending entries and
their candidates) and give the job-def its result schema and decoder.
The check before the result is stored keeps validating the same schema.

Closes #12339

AI-assisted-by: deepseek-v4.1-flash
Signed-off-by: Andrey Antukh <niwi@niwi.nz>

* 🐛 Take the imported file id from the entry, not the map key

The wizard looked a file up in the selection with the key of the
resolution map while it had indexed the selection by the :id of the
entry. The key is keywordized on its way through the job result, so the
lookup missed and the confirm button created no library link at all,
with no error and with the modal closing as if it had worked.

Take the id from the entry, the source the rest of the wizard already
uses, and move the decision to a pure function covered by tests of both
shapes of the resolution.

Relates to #12339

AI-assisted-by: deepseek-v4.1-flash
Signed-off-by: Andrey Antukh <niwi@niwi.nz>

* 🐛 Test that an import result keeps its ids as uuids

The regression this branch fixes is only visible through the read path
of the substrate: the resolution of an import is a map keyed by the new
file id, and a test that works with the value in memory never sees the
keys coming back as keywords from the plain JSON column.

Import a package whose library needs a replacement chosen by the user,
read the result with `jobs/get-job` as the client does, and check the
keys, the file ids and the ids of the candidates. Without the result
decoder of the import job-def the test answers with text ids.

Relates to #12339

AI-assisted-by: deepseek-v4.1-flash
Signed-off-by: Andrey Antukh <niwi@niwi.nz>

* 📚 Document the result schema of a job-def

The memories and the docstring of the substrate said a job result has no
schema of its own and comes back as the JSON the row holds. A job-def can
now declare its result schema and decoder, and the substrate applies them
when the row is read.

Keep the rule next to it: a result map keyed by uuid needs its schema
declared, or every key comes back as a keyword, and the link-later step
of the import wizard takes the file id from the entry and never from the
key of the resolution map.

Relates to #12339

AI-assisted-by: deepseek-v4.1-flash
Signed-off-by: Andrey Antukh <niwi@niwi.nz>

* 📚 Fix the documented contract of a job result

The docstring of get-job said a result has no schema of its own and comes
back as the JSON the row holds, which is what this branch changes: the
result is decoded with the schema its job-def declares, and a result that
does not conform never reaches the column because the write validates it.

Say the same in the management command that stores the result of an
external worker, the other writer of the column.

Relates to #12339

AI-assisted-by: deepseek-v4.1-flash
Signed-off-by: Andrey Antukh <niwi@niwi.nz>

* 🐛 Check the result of a job against its schema before storing it

The result of a job crosses a plain JSON column, so a uuid comes back as
text and a uuid used as a map key as a keyword. The job-def now declares
the schema of its result and the substrate compiles the decoder from it,
which is what the read path uses to get the types back.

The same schema is the contract of the write path: complete refuses a
result that does not conform, with the malli explanation, so nothing
invalid reaches the column, and the answer is the same for the runner and
for the external worker that completes a job through the management RPC.
A job that is no longer completable is skipped without checking, and the
read path decodes only: a row written by another version stays readable.

Relates to #12339

AI-assisted-by: deepseek-v4.1-flash
Signed-off-by: Andrey Antukh <niwi@niwi.nz>

* 🐛 Check the result of an import inside its transaction

The check ran after the transaction of the import had committed, so a
result the schema does not accept failed the job with the files already
created in the destination project: a half imported project and an error.
The core decides whether the whole import is kept or rolled back, so the
check belongs inside its transaction, which is what turns a contract
violation into a rollback.

The substrate checks the result again when the job is completed, which is
the only net for a job that does not validate its own result.

Relates to #12339

AI-assisted-by: deepseek-v4.1-flash
Signed-off-by: Andrey Antukh <niwi@niwi.nz>

* 🐛 Cover the edges of an import result read by the client

The shape of a result with no libraries to resolve, and the one of a
version 1 import, is the most common in production and neither was covered
by the tests that read a result the way the client does.

Cover both through jobs/get-job, and the case of a registry that no longer
knows the kind of the row: it answers with the JSON of the column,
keywordized keys and uuid as text, instead of raising. Say in the two tests
that depend on the decoder what implementation without it makes them fail.

Relates to #12339

AI-assisted-by: deepseek-v4.1-flash
Signed-off-by: Andrey Antukh <niwi@niwi.nz>

* 📚 Fix the prose and the fixtures of the result contract

The result schema of a job-def is what the read decodes with and what the
write checks: say that in the memories, with the rule that any uuid of a
result needs it, whether a value or a map key.

The motive of the wizard helper was a fact this branch makes false: what a
key of the resolution map holds depends on the result schema of the job-def
that wrote it, so the rule is to read the id from the entry. Same
correction in its memory and in the test that pins the shape.

The candidates of a pending entry match by slugified name, not by name, and
every test row now states the kind of its job, because the name is what
resolves its job-def when the result is read or checked.

Relates to #12339

AI-assisted-by: deepseek-v4.1-flash
Signed-off-by: Andrey Antukh <niwi@niwi.nz>

* ♻️ Keep the wizard link helper private and cover its shape

The helper that decides which links the wizard creates belongs to its
namespace, as its neighbours do: reach it from the test with @#', the
pattern the frontend tests already use for a private helper.

Cover the shapes it had no test for: a file with two libraries pending
resolution, a file whose libraries were auto-linked, and a selection where
only one of two entries was resolved.

Relates to #12339

AI-assisted-by: deepseek-v4.1-flash
Signed-off-by: Andrey Antukh <niwi@niwi.nz>

* ♻️ Share one job-row helper across the job tests

Three test namespaces had their own copy of the same insert, differing only
in the kind of the job and in whether the row holds the resource it
consumes. Move it to the test helpers as create-job*, with the kind of the
job as its first argument: it is what resolves the job-def of the row when
its result is read or checked, so every test states it.

The bang of the name goes away with it: the design rules keep it for the
mutating API of app.db, and this is a fixture of the tests. The three
namespaces lose the requires that only their removed copy used.

Relates to #12339

AI-assisted-by: deepseek-v4.1-flash
Signed-off-by: Andrey Antukh <niwi@niwi.nz>

---------

Signed-off-by: Andrey Antukh <niwi@niwi.nz>
2026-10-15 19:18:04 +02:00

1783 lines
83 KiB
Clojure
Raw 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.jobs-test
(:require
[app.common.exceptions :as ex]
[app.common.schema :as sm]
[app.common.time :as ct]
[app.common.uuid :as uuid]
[app.config :as cf]
[app.db :as db]
[app.jobs :as jobs]
[app.metrics :as-alias mtx]
[app.msgbus :as mbus]
[backend-tests.helpers :as th]
[clojure.test :as t]
[cuerdas.core :as str]
[integrant.core :as ig]))
(t/use-fixtures :once th/state-init)
(defn- test-fixture [next]
(th/database-reset next))
(t/use-fixtures :each test-fixture)
(defn- raised
"Run `f` and answer with the exception it raised, or nil when it did not."
[f]
(try (f) nil (catch Throwable cause cause)))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; JOB-DEF (plain handler + precompiled init-key)
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
(defn echo-handler
"Plain handler function, importable and testable without integrant.
Dual behavior: when invoked in-process (no job-id in cfg), heartbeat/progress
are no-ops. When invoked via the runner (with job-id in cfg), they write to
the job row. This allows the same handler to be tested both ways."
[cfg params]
(when (::jobs/job-id cfg)
(jobs/heartbeat cfg)
(jobs/heartbeat cfg :progress {:stage :half
:counters {:work {:current 1 :total 2}}}))
params)
(def schema:echo-params
[:map
[:text ::sm/text]
[:object :keyword]
[:deleted-at ::ct/inst]
[:id ::sm/uuid]
[:file-id {:optional true} ::sm/uuid]])
(defmethod ig/init-key ::echo-job
[_ cfg]
{::jobs/name :echo
::jobs/schema schema:echo-params
::jobs/handler (fn [_context params]
(echo-handler cfg params))
::jobs/decoder (sm/decoder schema:echo-params sm/json-transformer)
::jobs/validator (sm/validator schema:echo-params)})
(def schema:typed-result
"A result with a uuid keyed map, the shape that needs a declared schema
to survive the plain JSON round trip of the `job.result` column."
[:map
[:file-ids [:vector ::sm/uuid]]
[:resolution [:map-of ::sm/uuid [:map [:id ::sm/uuid]]]]])
(defmethod ig/init-key ::typed-job
[_ cfg]
{::jobs/name :typed
::jobs/schema schema:echo-params
::jobs/handler (fn [_context params]
(echo-handler cfg params))
::jobs/decoder (sm/decoder schema:echo-params sm/json-transformer)
::jobs/validator (sm/validator schema:echo-params)
;; the substrate compiles the decoder from this schema and checks the
;; result against it before storing it
::jobs/result-schema schema:typed-result})
(defn- get-job-defs
[]
(-> {::jobs/defs {:echo (ig/ref ::echo-job)
:typed (ig/ref ::typed-job)}
::echo-job {::db/pool th/*pool*}
::typed-job {::db/pool th/*pool*}}
(ig/expand)
(ig/init)
(get ::jobs/defs)))
(defn- make-cfg
[defs]
{::jobs/defs defs
::db/pool th/*pool*
::mtx/metrics (get th/*system* :app.metrics/metrics)})
(defn- make-params
[]
{:text "hello"
:object :snapshot
:deleted-at (ct/now)
:id (uuid/next)
:file-id (uuid/next)})
(defn- insert-job-statement?
"True when a `db/exec-one!` call is the job INSERT, whatever the
argument shape (a `[sql & params]` vector or a bare SQL string)."
[args]
(let [stmt (second args)
sql (if (string? stmt) stmt (first stmt))]
(and (string? sql) (str/starts-with? sql "insert into job"))))
(def ^:private test-error
{:type :internal
:code :boom
:hint "boom"})
(defn- get-progresss
"Progress events of a job, oldest first, with decoded payloads."
[job-id]
(->> (th/db-exec! ["SELECT payload FROM job_event
WHERE job_id = ? AND kind = 'progress'
ORDER BY created_at ASC, id ASC"
job-id])
(mapv #(jobs/decode-progress (:payload %)))))
(defn- get-events
"Every event of a job, in insertion order."
[job-id]
(->> (th/db-exec! ["SELECT kind, payload FROM job_event
WHERE job_id = ? ORDER BY id ASC" job-id])
(mapv (fn [{:keys [kind payload]}]
{:kind kind :payload (db/decode-json-pgobject payload)}))))
(defn- get-outcomes
"Outcomes of the end events of a job, oldest first."
[job-id]
(mapv (comp :outcome :payload) (filter #(= "end" (:kind %)) (get-events job-id))))
(defn- get-kinds
[job-id]
(mapv :kind (get-events job-id)))
(defn- fake-msgbus
"A minimal msgbus that records every publication on the atom."
[messages]
(reify mbus/IMsgBus
(-sub [_ _ _] nil)
(-pub [_ topic message]
(swap! messages conj {:topic topic :message message})
nil)
(-purge [_ _] nil)))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; TESTS
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
(t/deftest submit-persists-optional-profile-and-resource-references
(let [cfg (make-cfg (get-job-defs))
profile-id (:id (th/create-profile* 1))
resource-id (uuid/next)]
;; the resource reference points to a real storage object: the column
;; is a foreign key kept for garbage collection
(th/db-insert! :storage-object {:id resource-id :backend "test"})
(let [internal-id (jobs/submit cfg {::jobs/name :echo
::jobs/params (make-params)})
owned-id (jobs/submit cfg {::jobs/name :echo
::jobs/params (make-params)
::jobs/profile-id profile-id
::jobs/resource-id resource-id})
internal-row (jobs/get-job cfg internal-id)
owned-row (jobs/get-job cfg owned-id)]
(t/testing "profile_id stays nullable for internal jobs"
(t/is (nil? (:profile-id internal-row)))
(t/is (nil? (:resource-id internal-row))))
(t/testing "an explicit profile is persisted as the owner of the job"
(t/is (= profile-id (:profile-id owned-row))))
(t/testing "the resource reference is a column, not part of params"
(t/is (= resource-id (:resource-id owned-row)))
(t/is (not (contains? (:params owned-row) :resource-id)))))))
(t/deftest submit-rejects-invalid-profile-and-resource-references
(let [cfg (make-cfg (get-job-defs))]
(t/testing "a non uuid profile-id is rejected before the insert"
(t/is (thrown? Exception
(jobs/submit cfg {::jobs/name :echo
::jobs/params (make-params)
::jobs/profile-id "not-a-uuid"}))))
(t/testing "a non uuid resource-id is rejected before the insert"
(t/is (thrown? Exception
(jobs/submit cfg {::jobs/name :echo
::jobs/params (make-params)
::jobs/resource-id "not-a-uuid"}))))
(t/testing "no job row was created by the rejected submits"
(t/is (= 0 (:cnt (th/db-exec-one! ["SELECT count(*) AS cnt FROM job"])))))))
(t/deftest submit-persists-an-expiry-date
(let [cfg (make-cfg (get-job-defs))
expires-at (ct/plus (ct/now) (ct/duration {:days 7}))]
(t/testing "an explicit expiry is persisted as the retention of the job"
(let [job-id (jobs/submit cfg {::jobs/name :echo
::jobs/params (make-params)
::jobs/expires-at expires-at})
row (jobs/get-job cfg job-id)]
(t/is (= (inst-ms expires-at) (inst-ms (:expires-at row))))))
(t/testing "without the option the column stays nil, as before"
(let [job-id (jobs/submit cfg {::jobs/name :echo
::jobs/params (make-params)})
row (jobs/get-job cfg job-id)]
(t/is (nil? (:expires-at row)))))))
(t/deftest submit-rejects-an-invalid-expiry-date
(let [cfg (make-cfg (get-job-defs))]
(t/testing "a value that is not an instant is rejected before the insert"
(t/is (thrown? Exception
(jobs/submit cfg {::jobs/name :echo
::jobs/params (make-params)
::jobs/expires-at "not-an-instant"}))))
(t/testing "no job row was created by the rejected submit"
(t/is (= 0 (:cnt (th/db-exec-one! ["SELECT count(*) AS cnt FROM job"])))))))
(t/deftest progress-report-validates-its-payload
(let [cfg (make-cfg (get-job-defs))
job-id (jobs/submit cfg {::jobs/name :echo
::jobs/params (make-params)})]
(t/testing "stage is mandatory"
(t/is (thrown? Exception
(jobs/heartbeat cfg :job-id job-id
:progress {:counters {:work {:current 1}}}))))
(t/testing "stage must be a keyword, not a label"
(t/is (thrown? Exception
(jobs/heartbeat cfg :job-id job-id
:progress {:stage "work"}))))
(t/testing "unknown keys are rejected, at the top and inside a counter"
(t/is (thrown? Exception
(jobs/heartbeat cfg :job-id job-id
:progress {:stage :work :step 1})))
(t/is (thrown? Exception
(jobs/heartbeat cfg :job-id job-id
:progress {:stage :work
:counters {:work {:current 1
:step 1}}}))))
(t/testing "a counter current must not be negative"
(t/is (thrown-with-msg? Exception #"negative"
(jobs/heartbeat cfg :job-id job-id
:progress {:stage :work
:counters {:work {:current -1}}}))))
(t/testing "a counter total must be positive"
(t/is (thrown-with-msg? Exception #"positive"
(jobs/heartbeat cfg :job-id job-id
:progress {:stage :work
:counters {:work {:current 0
:total 0}}}))))
(t/testing "a counter current must not be greater than its total"
(t/is (thrown-with-msg? Exception #"not lower than current"
(jobs/heartbeat cfg :job-id job-id
:progress {:stage :work
:counters {:work {:current 5
:total 4}}}))))
(t/testing "keys are limited to a name and not a payload"
(t/is (thrown-with-msg? Exception #"64"
(jobs/heartbeat cfg :job-id job-id
:progress {:stage (keyword (apply str (repeat 65 "x")))}))))
(t/testing "a milestone without counters is accepted"
(t/is (= 2 (jobs/heartbeat cfg :job-id job-id :progress {:stage :relations}))))
(t/testing "a milestone with counters is accepted and stored as it is"
(swap! jobs/progresses dissoc job-id)
(swap! jobs/heartbeats dissoc job-id)
(t/is (= 2 (jobs/heartbeat cfg :job-id job-id
:progress {:stage :pages
:counters {:files {:current 2 :total 5}
:pages {:current 3 :total 8}}})))
(t/is (= [{:stage :relations}
{:stage :pages
:counters {:files {:current 2 :total 5}
:pages {:current 3 :total 8}}}]
(get-progresss job-id))))
(t/testing "a rejected report never reaches the durable log"
(t/is (= 2 (count (get-progresss job-id)))))))
(t/deftest progress-event-publishes-msgbus-for-profile-jobs
(let [cfg (make-cfg (get-job-defs))
profile-id (:id (th/create-profile* 1))
messages (atom [])
job-cfg (assoc cfg ::mbus/msgbus (fake-msgbus messages))
job-id (jobs/submit job-cfg {::jobs/name :echo
::jobs/params (make-params)
::jobs/profile-id profile-id})]
(t/is (pos? (jobs/heartbeat job-cfg :job-id job-id
:progress {:stage :pages
:counters {:files {:current 3 :total 7}}})))
(t/testing "the event is published on the topic of the job profile"
(t/is (= 1 (count @messages)))
(let [{:keys [topic message]} (first @messages)]
(t/is (= profile-id topic))
(t/is (= :job-event (:type message)))
(t/is (= job-id (:job-id message)))
(t/is (= profile-id (:profile-id message)))
(t/is (= "progress" (:kind message)))
(t/is (= {:stage :pages
:counters {:files {:current 3 :total 7}}}
(:payload message)))
(t/is (some? (:event-id message)))
(t/is (some? (:created-at message)))))))
(t/deftest progress-event-of-internal-job-is-not-published
(let [cfg (make-cfg (get-job-defs))
messages (atom [])
job-cfg (assoc cfg ::mbus/msgbus (fake-msgbus messages))
job-id (jobs/submit job-cfg {::jobs/name :echo
::jobs/params (make-params)})]
(t/is (pos? (jobs/heartbeat job-cfg :job-id job-id :progress {:stage :work})))
(t/testing "the event is stored but nothing is published"
(t/is (= 1 (count (get-progresss job-id))))
(t/is (= [] @messages)))))
(t/deftest progress-event-of-profile-job-requires-msgbus
(let [cfg (make-cfg (get-job-defs))
profile-id (:id (th/create-profile* 1))
job-id (jobs/submit cfg {::jobs/name :echo
::jobs/params (make-params)
::jobs/profile-id profile-id})]
(t/testing "the forced route propagates the missing msgbus"
(t/is (thrown-with-msg? Exception #"require ::mbus/msgbus"
(jobs/heartbeat cfg
:job-id job-id
:progress {:stage :work}
::jobs/force? true)))
(t/is (= [] (get-progresss job-id))
"the event must not be stored without its notification"))
(t/testing "the handler route logs the failure and keeps the job alive"
(th/db-update! :job {:modified-at (ct/in-past {:minutes 5})} {:id job-id})
(swap! jobs/heartbeats dissoc job-id)
(t/is (= 1 (jobs/heartbeat cfg :job-id job-id
:progress {:stage :work})))
(t/is (= [] (get-progresss job-id))))))
(t/deftest submit-validates-params-with-job-schema
(let [cfg (make-cfg (get-job-defs))]
(t/is (thrown-with-msg? Exception #"check error"
(jobs/submit cfg {::jobs/name :echo
::jobs/params {:text "hello"
:object :snapshot
:deleted-at (ct/now)
:id (uuid/next)
:file-id "not-an-uuid"}}))))
(t/testing "missing job definition raises a clear error"
(let [cfg (make-cfg (get-job-defs))]
(t/is (thrown-with-msg? Exception #"no job definition"
(jobs/submit cfg {::jobs/name :unknown
::jobs/params {}})))))
(t/testing "the job name is a keyword, and only a keyword"
(let [cfg (make-cfg (get-job-defs))]
(t/is (thrown? clojure.lang.ExceptionInfo
(jobs/submit cfg {::jobs/name "echo"
::jobs/params {}})))
(t/is (thrown? clojure.lang.ExceptionInfo
(jobs/invoke (assoc cfg
::jobs/name "echo"
::jobs/params {})))))))
(t/deftest submit-persists-row-with-json-params
(let [cfg (make-cfg (get-job-defs))
params (make-params)
job-id (jobs/submit cfg {::jobs/name :echo
::jobs/params params
::jobs/queue :test
::jobs/delay 1000
::jobs/priority 200
::jobs/label "test-label"
::jobs/max-retries 5})
row (jobs/get-job cfg job-id)]
(t/is (uuid? job-id))
(t/is (= "echo" (:name row)))
(t/is (= (cf/get :tenant) (:tenant row)))
(t/is (= "test" (:queue row)))
(t/is (= "test-label" (:label row)))
(t/is (= 200 (:priority row)))
(t/is (= 5 (:max-retries row)))
(t/is (= "new" (:status row)))
(t/testing "scheduled_at respects the submitted delay"
(let [delay (ct/duration {:seconds 1})]
(t/is (pos? (- (inst-ms (:scheduled-at row)) (inst-ms (ct/now))))
(str "scheduled-at should be in the future: "
(pr-str (:scheduled-at row))))))
(t/testing "the row exposes params and no legacy columns"
(t/is (some? (:params row)))
(t/is (not (contains? row :props)))
(t/is (not (contains? row :progress)))
(t/is (not (contains? row :target))))
(t/testing "params are stored as plain JSON (no transit tags) and decoded"
(let [params (:params row)]
(t/is (map? params))
(t/is (nil? (some-> (th/db-exec-one! ["SELECT params::text FROM job WHERE id = ?"
job-id])
:params
(str/index-of "~#"))))
(t/testing "round-trip: decoded values have proper clojure types"
(let [decoded (jobs/decode-params (jobs/get-job-def (get cfg ::jobs/defs) :echo)
(:params row))]
(t/is (keyword? (:object decoded)))
(t/is (= :snapshot (:object decoded)))
(t/is (uuid? (:id decoded)))
(t/is (uuid? (:file-id decoded)))
(t/is (ct/inst? (:deleted-at decoded)))))))))
(t/deftest submit-dedupe-replaces-not-due-new-rows
(let [cfg (make-cfg (get-job-defs))
params (make-params)
label "dedupe-label"
id1 (jobs/submit cfg {::jobs/name :echo
::jobs/params params
::jobs/label label
::jobs/dedupe true
::jobs/delay 10000})
id2 (jobs/submit cfg {::jobs/name :echo
::jobs/params params
::jobs/label label
::jobs/dedupe true
::jobs/delay 10000})]
(t/is (uuid? id1))
(t/is (uuid? id2))
(t/is (not= id1 id2))
(t/testing "the first new row (not due yet) is removed"
(t/is (nil? (jobs/get-job cfg id1)))
(t/testing "and only the new row remains"
(t/is (= 1 (:cnt (th/db-exec-one! ["SELECT count(*) AS cnt FROM job
WHERE label = ?" label]))))))))
(t/deftest submit-dedupe-keeps-due-and-running-rows
(let [cfg (make-cfg (get-job-defs))
params (make-params)
label "dedupe-active-label"
id1 (jobs/submit cfg {::jobs/name :echo
::jobs/params params
::jobs/label label})
_ (th/db-update! :job {:status "running"} {:id id1})
id2 (jobs/submit cfg {::jobs/name :echo
::jobs/params params
::jobs/label label
::jobs/dedupe true})]
(t/is (some? (jobs/get-job cfg id1)))
(t/is (= 2 (:cnt (th/db-exec-one! ["SELECT count(*) AS cnt FROM job
WHERE label = ?" label])))))
(let [cfg (make-cfg (get-job-defs))
params (make-params)
label "dedupe-due-label"
id1 (jobs/submit cfg {::jobs/name :echo
::jobs/params params
::jobs/label label})
_ (th/db-update! :job {:scheduled-at (ct/minus (ct/now)
(ct/duration {:seconds 5}))}
{:id id1})
id2 (jobs/submit cfg {::jobs/name :echo
::jobs/params params
::jobs/label label
::jobs/dedupe true})]
(t/is (some? (jobs/get-job cfg id1)))
(t/is (= 2 (:cnt (th/db-exec-one! ["SELECT count(*) AS cnt FROM job
WHERE label = ?" label]))))))
(t/deftest submit-dedupe-leaves-another-tenants-rows-alone
;; The DELETE matches name, queue and label. Without the tenant in the
;; match it would reach into another instance's not-started jobs.
(let [cfg (make-cfg (get-job-defs))
params (make-params)
label "dedupe-shared-label"
foreign-id (uuid/next)]
(th/db-insert! :job {:id foreign-id
:name "echo"
:tenant "other-tenant"
:queue "default"
:label label
:params (db/json params)
:status "new"
:priority 100
:max-retries 3
:retry-num 0
:scheduled-at (ct/plus (ct/now) (ct/duration {:seconds 10}))
:created-at (ct/now)
:modified-at (ct/now)})
(jobs/submit cfg {::jobs/name :echo
::jobs/params params
::jobs/queue :default
::jobs/label label
::jobs/dedupe true
::jobs/delay 10000})
(t/is (some? (jobs/get-job cfg foreign-id))
"another tenant's matching job must survive our dedupe")
(t/testing "and only our own row was written"
(t/is (= 1 (:cnt (th/db-exec-one! ["SELECT count(*) AS cnt FROM job
WHERE label = ?
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)
opts {::jobs/name :echo
::jobs/params params
::jobs/dedupe true
::jobs/label "atomic-label"}
kept (jobs/submit cfg opts)
orig @#'db/exec-one!]
;; fault the INSERT: the DELETE before it must roll back too
(alter-var-root #'db/exec-one!
(constantly (fn [& args]
(when (insert-job-statement? args)
(throw (ex-info "boom" {})))
(apply orig args))))
(try
(t/is (thrown? Exception (jobs/submit cfg opts)))
(t/testing "the original row survives, no duplicate left behind"
(t/is (some? (jobs/get-job cfg kept)))
(t/is (= 1 (:cnt (th/db-exec-one! ["SELECT count(*) AS cnt FROM job WHERE label = ?" "atomic-label"])))))
(finally
(alter-var-root #'db/exec-one! (constantly orig))))))
(t/deftest submit-dedupe-atomic-on-autocommit-conn
(let [cfg (make-cfg (get-job-defs))
params (make-params)
opts {::jobs/name :echo
::jobs/params params
::jobs/dedupe true
::jobs/label "atomic-autocommit-label"}
kept (jobs/submit cfg opts)
orig @#'db/exec-one!]
;; a raw connection outside any transaction: DELETE+INSERT must
;; still share one transaction opened on it
(with-open [conn (db/open th/*pool*)]
(alter-var-root #'db/exec-one!
(constantly (fn [& args]
(when (insert-job-statement? args)
(throw (ex-info "boom" {})))
(apply orig args))))
(try
(t/is (thrown? Exception
(jobs/submit (assoc cfg ::db/conn conn) opts)))
(t/testing "the original row survives, no duplicate left behind"
(t/is (some? (jobs/get-job cfg kept)))
(t/is (= 1 (:cnt (th/db-exec-one! ["SELECT count(*) AS cnt FROM job WHERE label = ?" "atomic-autocommit-label"])))))
(finally
(alter-var-root #'db/exec-one! (constantly orig)))))))
(t/deftest submit-requires-metrics-on-bare-connectable
(let [defs (get-job-defs)
prev @@#'jobs/defs-registry]
(try
(reset! @#'jobs/defs-registry defs)
(t/is (thrown-with-msg? Exception #"missing ::mtx/metrics"
(jobs/submit th/*pool* {::jobs/name :echo
::jobs/params (make-params)})))
(finally
(reset! @#'jobs/defs-registry prev)))))
(t/deftest plain-handler-is-testable-without-integrant
(let [params (make-params)]
(t/is (= params (echo-handler {} params))))
(let [cfg (make-cfg (get-job-defs))
params (make-params)
job-id (jobs/submit cfg {::jobs/name :echo
::jobs/params params})]
(t/testing "handler receives cfg with job-id context for heartbeats"
(t/is (= params (echo-handler (assoc cfg ::jobs/job-id job-id)
params))))))
(t/deftest submit-strips-rollback-testing-flag-from-params
(let [cfg (make-cfg (get-job-defs))
params (assoc (make-params) :rollback? true)
job-id (jobs/submit cfg {::jobs/name :echo
::jobs/params params})]
(t/testing "durable rows never carry the in-process escape hatch"
(t/is (nil? (:rollback? (:params (jobs/get-job cfg job-id))))))))
(t/deftest heartbeat-respects-throttle
(let [cfg (make-cfg (get-job-defs))
job-id (jobs/submit cfg {::jobs/name :echo
::jobs/params (make-params)})]
(t/testing "first heartbeat writes"
;; Backdate modified-at so the beat is unambiguous: the assertion
;; compares against the value we set, not against created-at, which
;; can land on the same millisecond.
(let [backdated (ct/in-past {:seconds 5})]
(th/db-update! :job {:modified-at backdated} {:id job-id})
(t/is (= 1 (jobs/heartbeat cfg :job-id job-id)))
(let [row (jobs/get-job cfg job-id)]
(t/is (> (inst-ms (:modified-at row)) (inst-ms backdated))))))
(t/testing "immediate second heartbeat does not write (throttled)"
(let [row1 (jobs/get-job cfg job-id)
_ (jobs/heartbeat cfg :job-id job-id)
row2 (jobs/get-job cfg job-id)]
(t/is (= (inst-ms (:modified-at row1))
(inst-ms (:modified-at row2))))))))
(t/deftest heartbeat-interrupts-on-terminal-states
(let [cfg (make-cfg (get-job-defs))
job-id (jobs/submit cfg {::jobs/name :echo
::jobs/params (make-params)})]
(t/testing "a terminal state raises the interrupt instead of touching the row"
(th/db-update! :job {:status "completed"
:modified-at (ct/in-past {:days 10})}
{:id job-id})
(swap! @#'jobs/heartbeats dissoc job-id)
(let [before (jobs/get-job cfg job-id)
cause (raised #(jobs/heartbeat cfg :job-id job-id))]
(t/is (= :interrupt (th/ex-type cause)))
(t/is (= :job-interrupted (th/ex-code cause)))
(t/is (= job-id (:job-id (ex-data cause))))
(t/is (= "completed" (:status (ex-data cause))))
(t/is (= (inst-ms (:modified-at before))
(inst-ms (:modified-at (jobs/get-job cfg job-id)))))))))
(t/deftest heartbeat-interrupts-when-the-row-is-gone
(let [cfg (make-cfg (get-job-defs))
job-id (jobs/submit cfg {::jobs/name :echo
::jobs/params (make-params)})]
(th/db-force-delete :job {:id job-id})
(let [cause (raised #(jobs/heartbeat cfg :job-id job-id))]
(t/is (= :interrupt (th/ex-type cause)))
(t/is (= :job-interrupted (th/ex-code cause)))
(t/is (= "gone" (:status (ex-data cause)))))))
(t/deftest a-transient-progress-failure-is-not-an-interrupt
"The handler path logs and ignores a progress insert that fails, but an
interrupt must not be swallowed with it: only an interrupt means the job
is gone."
(let [cfg (make-cfg (get-job-defs))
job-id (jobs/submit cfg {::jobs/name :echo
::jobs/params (make-params)})]
(swap! jobs/progresses dissoc job-id)
(swap! jobs/heartbeats dissoc job-id)
;; the insert is the boundary that fails: the job itself is active, so
;; the beat answers with the touch it did write instead of raising
(with-redefs-fn {#'jobs/report-progress (fn [& _] (throw (ex-info "boom" {})))}
#(t/is (= 1 (jobs/heartbeat cfg :job-id job-id :progress {:stage :work}))))
(t/is (= [] (get-progresss job-id)))))
(t/deftest progress-events-respect-throttle-and-interrupt-on-terminal-states
(let [cfg (make-cfg (get-job-defs))
job-id (jobs/submit cfg {::jobs/name :echo
::jobs/params (make-params)})]
(t/testing "first progress report appends a progress event"
(t/is (= 2 (jobs/heartbeat cfg :job-id job-id
:progress {:stage :pages
:counters {:pages {:current 1 :total 10}}})))
(t/is (= [{:stage :pages :counters {:pages {:current 1 :total 10}}}]
(get-progresss job-id))))
(t/testing "immediate second progress report is throttled"
(t/is (= 0 (jobs/heartbeat cfg :job-id job-id
:progress {:stage :pages
:counters {:pages {:current 2 :total 10}}})))
(t/is (= [{:stage :pages :counters {:pages {:current 1 :total 10}}}]
(get-progresss job-id)))
(t/testing "after the throttle window elapses it appends again"
(swap! @#'jobs/progresses
(fn [m]
(update-in m [job-id]
#(ct/minus %
(ct/duration {:seconds 2})))))
(jobs/heartbeat cfg :job-id job-id
:progress {:stage :pages
:counters {:pages {:current 3 :total 10}}})
(t/is (= [{:stage :pages :counters {:pages {:current 1 :total 10}}}
{:stage :pages :counters {:pages {:current 3 :total 10}}}]
(get-progresss job-id))))))
(let [cfg (make-cfg (get-job-defs))
job-id (jobs/submit cfg {::jobs/name :echo
::jobs/params (make-params)})]
(t/testing "terminal states never get a progress event"
(th/db-update! :job {:status "completed"} {:id job-id})
(swap! @#'jobs/progresses dissoc job-id)
(swap! @#'jobs/heartbeats dissoc job-id)
(let [cause (raised #(jobs/heartbeat cfg :job-id job-id
:progress {:stage :work}))]
(t/is (= :interrupt (th/ex-type cause)))
(t/is (= [] (get-progresss job-id)))))))
(t/deftest throttle-prune-removes-stale-entries-keeps-fresh
"When the throttle map exceeds prune-threshold, stale entries (older than
prune-window) are removed and fresh entries are kept."
(let [cfg (make-cfg (get-job-defs))
job-id-1 (jobs/submit cfg {::jobs/name :echo
::jobs/params (make-params)})
job-id-2 (jobs/submit cfg {::jobs/name :echo
::jobs/params (make-params)})
now (ct/now)
stale-time (ct/minus now (ct/duration {:hours 2})) ;; older than 1h window
fresh-time (ct/minus now (ct/duration {:minutes 30}))] ;; within 1h window
;; Fill the heartbeats map past the threshold with one stale and one fresh entry
(with-redefs [jobs/prune-threshold 0] ;; force prune on next call
(reset! @#'jobs/heartbeats
{job-id-1 stale-time
job-id-2 fresh-time})
;; Trigger a heartbeat for a new job (should trigger prune)
(let [job-id-3 (jobs/submit cfg {::jobs/name :echo
::jobs/params (make-params)})]
(jobs/heartbeat cfg :job-id job-id-3)
;; After prune: stale entry (job-id-1) should be gone, fresh (job-id-2) should remain
(let [state @jobs/heartbeats]
(t/is (not (contains? state job-id-1)) "stale entry removed")
(t/is (contains? state job-id-2) "fresh entry kept")
(t/is (contains? state job-id-3) "new entry added"))))))
(t/deftest cancel-skips-terminal-jobs
(let [cfg (make-cfg (get-job-defs))
job-id (jobs/submit cfg {::jobs/name :echo
::jobs/params (make-params)})]
(t/testing "pending job can be cancelled"
(t/is (= 1 (jobs/cancel cfg job-id)))
(t/is (= "cancelled" (:status (jobs/get-job cfg job-id))))
(t/is (= ["cancelled"] (get-outcomes job-id)))
(t/testing "already cancelled job is not affected again"
(t/is (zero? (jobs/cancel cfg job-id)))
(t/is (= ["cancelled"] (get-outcomes job-id))))))
(doseq [status ["completed" "failed"]]
(t/testing (str status " job cannot be cancelled")
(let [cfg (make-cfg (get-job-defs))
job-id (jobs/submit cfg {::jobs/name :echo
::jobs/params (make-params)})]
(th/db-update! :job {:status status} {:id job-id})
(t/is (zero? (jobs/cancel cfg job-id)))
(t/is (= status (:status (jobs/get-job cfg job-id))))
(t/is (= [] (get-outcomes job-id)))))))
(t/deftest cancel-affects-pending-and-running-jobs
(let [cfg (make-cfg (get-job-defs))]
(doseq [status ["scheduled" "retry" "running"]]
(t/testing (str status " job can be cancelled")
(let [job-id (jobs/submit cfg {::jobs/name :echo
::jobs/params (make-params)})]
(th/db-update! :job {:status status} {:id job-id})
(t/is (= 1 (jobs/cancel cfg job-id)))
(t/is (= "cancelled" (:status (jobs/get-job cfg job-id))))
(t/is (= ["cancelled"] (get-outcomes job-id))))))))
(t/deftest get-user-status-maps-internal-statuses
(t/are [status expected] (= expected (jobs/get-user-status status))
"new" "pending"
"scheduled" "pending"
"retry" "pending"
:new "pending"
"running" "running"
"completed" "completed"
"failed" "failed"
"cancelled" "failed"
"aborted" "failed"))
(t/deftest job-defs-registry-validates-definitions
(t/testing "valid job-def map passes the assert"
(t/is (some? (get-job-defs))))
(t/testing "registry name mismatch is detected"
(t/is (thrown? Exception
(ig/init {::jobs/defs {::echo-job
{::jobs/name :other
::jobs/schema schema:echo-params
::jobs/handler (fn [_context params]
(echo-handler nil params))
::jobs/decoder (sm/decoder schema:echo-params sm/json-transformer)
::jobs/validator (sm/validator schema:echo-params)}}}))))
(t/testing "missing decoder/validator is detected"
(t/is (thrown? Exception
(ig/init {::jobs/defs {::echo-job
{::jobs/name :echo
::jobs/schema schema:echo-params
::jobs/handler (fn [_context params]
params)}}})))))
(t/deftest job-def-carries-optional-family-and-resource-role
(let [job-def {::jobs/name :echo
::jobs/family :export
::jobs/resource-role :output
::jobs/schema schema:echo-params
::jobs/handler (fn [_context params] params)
::jobs/decoder (sm/decoder schema:echo-params sm/json-transformer)
::jobs/validator (sm/validator schema:echo-params)}]
(t/testing "the metadata is accepted and readable from the registry"
(let [defs (-> (ig/init {::jobs/defs {:echo job-def}})
(get ::jobs/defs))
echo (jobs/get-job-def defs :echo)]
(t/is (= :export (::jobs/family echo)))
(t/is (= :output (::jobs/resource-role echo)))))
(t/testing "a family is a keyword, not a free string"
(t/is (thrown? Exception
(ig/init {::jobs/defs {:echo (assoc job-def
::jobs/family "export")}}))))
(t/testing "a resource role is a keyword too"
(t/is (thrown? Exception
(ig/init {::jobs/defs {:echo (assoc job-def
::jobs/resource-role "output")}}))))))
(t/deftest generic-schema-round-trip-preserves-type-sensitive-fields
"For each registered job-def, verify that type-sensitive fields (uuids, insts)
survive the JSON round-trip (db/json → decode). This catches the F2 class of
bug where uuid types are lost during the transit→JSON switch."
(let [defs (get-job-defs)]
(t/is (pos? (count defs)) "should have at least one job-def")
;; Test the echo job-def which we know exists in the test registry
(t/testing "echo job-def preserves types through JSON round-trip"
(let [echo-def (get defs :echo)
_ (t/is (some? echo-def) "echo job-def should exist")
;; Create a representative params map with type-sensitive fields
sample-params {:text "test"
:object :snapshot
:deleted-at (ct/now)
:id (uuid/random)
:file-id (uuid/random)}
;; Round-trip through JSON (simulates the job table storage)
;; db/json returns a PGobject, which is what decode-params expects
pg-obj (db/json sample-params)
decoded (jobs/decode-params echo-def pg-obj)]
;; Verify type-sensitive fields survived
(t/is (uuid? (:id decoded)) "id should be uuid after decode")
(t/is (uuid? (:file-id decoded)) "file-id should be uuid after decode")
(t/is (inst? (:deleted-at decoded)) "deleted-at should be inst after decode")))))
(t/deftest invoke-falls-back-to-global-registry
"Verify that invoke! uses the global registry fallback when ::defs is not
on the cfg, consistent with submit!."
;; Rebind the private registry atom instead of overwriting global state:
;; it auto-restores on exit, so no later test can observe it. Note
;; get-job-defs itself populates the global via ig/init (established
;; pattern, also used by the sibling tests); the snapshot below is
;; taken after that, so it pins exactly this test's rebinding.
(let [defs (get-job-defs)
before @@#'jobs/defs-registry]
(with-redefs [jobs/defs-registry (atom defs)]
;; cfg WITHOUT ::jobs/defs to exercise the fallback; the echo
;; job-def is known to exist in the rebound registry
(let [cfg {::db/pool th/*pool*}
result (jobs/invoke (assoc cfg ::jobs/name :echo
::jobs/params (make-params)))]
;; Should not throw; should find the job-def via the fallback.
;; The result is the params map (echo-handler returns params).
(t/is (some? result))))
(t/testing "global registry untouched"
(t/is (= before @@#'jobs/defs-registry)))))
(t/deftest heartbeat-and-progress-noop-when-job-id-nil
"When *job-id* is nil (in-process invoke! without a job row),
heartbeat! and progress! are no-ops — they do not write to the
database or update the throttle atoms."
(let [cfg (make-cfg (get-job-defs))
job-id (jobs/submit cfg {::jobs/name :echo
::jobs/params (make-params)})]
(t/testing "heartbeat! is a no-op when *job-id* is nil"
(let [row-before (jobs/get-job cfg job-id)]
(binding [jobs/*job-id* nil]
(jobs/heartbeat cfg))
(let [row-after (jobs/get-job cfg job-id)]
(t/is (= (inst-ms (:modified-at row-before))
(inst-ms (:modified-at row-after)))
"modified-at should not change"))))
(t/testing "progress reporting is a no-op when *job-id* is nil"
(binding [jobs/*job-id* nil]
(t/is (nil? (jobs/heartbeat cfg :progress {:stage :work}))))
(t/is (= [] (get-progresss job-id))
"no progress event should be stored"))))
(t/deftest progress-interrupts-on-terminal-states
"A progress report never stores an event when the job is already in a
terminal state (completed, failed, cancelled, aborted): the row is locked and
checked before the insert, so a report can never race a terminal
transition, and the beat raises the interrupt instead."
(let [cfg (make-cfg (get-job-defs))]
(doseq [status ["completed" "failed" "cancelled" "aborted"]]
(t/testing (str "progress interrupts on " status " status")
(let [job-id (jobs/submit cfg {::jobs/name :echo
::jobs/params (make-params)})]
(th/db-update! :job {:status status} {:id job-id})
;; reset the throttles so should-write? would allow the writes
(swap! jobs/progresses dissoc job-id)
(swap! jobs/heartbeats dissoc job-id)
(binding [jobs/*job-id* job-id]
(let [cause (raised #(jobs/heartbeat cfg :progress {:stage :work}))]
(t/is (= :interrupt (th/ex-type cause)))
(t/is (= status (:status (ex-data cause))))))
(t/is (= [] (get-progresss job-id))))))))
(t/deftest defs-halt-clears-module-registry
(let [prev @@#'jobs/defs-registry]
;; init populates the module registry
(get-job-defs)
(t/is (contains? @@#'jobs/defs-registry :echo))
(try
(ig/halt-key! ::jobs/defs nil)
(t/is (= {} @@#'jobs/defs-registry))
(finally
(reset! @#'jobs/defs-registry prev)))))
(t/deftest heartbeat-bypasses-caller-transaction
(let [cfg (make-cfg (get-job-defs))
job-id (jobs/submit cfg {::jobs/name :echo
::jobs/params (make-params)})]
;; backdate so the beat is strictly greater (no same-millis flake)
(th/db-update! :job {:modified-at (ct/in-past {:minutes 5})} {:id job-id})
;; beat inside a transaction that is rolled back: the beat must
;; still be visible (it went through the pool, not the tx)
(db/tx-run! (assoc cfg ::db/rollback true)
(fn [{:keys [::db/conn]}]
(jobs/heartbeat (assoc cfg ::db/conn conn) :job-id job-id)))
(let [row (jobs/get-job cfg job-id)]
(t/is (> (inst-ms (:modified-at row))
(inst-ms (ct/in-past {:minutes 5})))))))
(t/deftest progress-bypasses-caller-transaction
(let [cfg (make-cfg (get-job-defs))
job-id (jobs/submit cfg {::jobs/name :echo
::jobs/params (make-params)})]
(db/tx-run! (assoc cfg ::db/rollback true)
(fn [{:keys [::db/conn]}]
(jobs/heartbeat (assoc cfg ::db/conn conn)
:job-id job-id
:progress {:stage :work})))
(t/is (= [{:stage :work}] (get-progresss job-id)))))
(t/deftest progress-notification-survives-a-rolled-back-caller
(let [cfg (make-cfg (get-job-defs))
profile-id (:id (th/create-profile* 1))
messages (atom [])
job-cfg (assoc cfg ::mbus/msgbus (fake-msgbus messages))
job-id (jobs/submit job-cfg {::jobs/name :echo
::jobs/params (make-params)
::jobs/profile-id profile-id})]
;; the report owns its transaction, so it commits and publishes even
;; though the transaction that triggered it rolls back
(db/tx-run! (assoc job-cfg ::db/rollback true)
(fn [{:keys [::db/conn]}]
(jobs/heartbeat (assoc job-cfg ::db/conn conn)
:job-id job-id
:progress {:stage :work})))
(t/testing "the event is stored"
(t/is (= [{:stage :work}] (get-progresss job-id))))
(t/testing "and the notification went out with its own commit"
(t/is (= 1 (count @messages)))
(t/is (= "progress" (:kind (:message (first @messages))))))))
(t/deftest heartbeat-requires-a-cfg-map
(let [cfg (make-cfg (get-job-defs))
job-id (jobs/submit cfg {::jobs/name :echo
::jobs/params (make-params)})]
(t/testing "a bare pool is a caller error, not something to accommodate"
(t/is (thrown? Exception
(jobs/heartbeat th/*pool*
:job-id job-id
:progress {:stage :work}))))))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; HANDLER CONTEXT
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
(defn- mk-row
"A job row as the runner and make-context see it."
[overrides]
(merge {:id (uuid/next)
:name "echo"
:tenant (cf/get :tenant)
:queue "default"
:label nil
:profile-id nil
:resource-id nil
:retry-num 0
:max-retries 3}
overrides))
(t/deftest make-context-selects-exactly-the-agreed-keys
(let [context (jobs/make-context (mk-row {}))]
(t/is (= #{:id :name :label :resource-id :profile-id} (set (keys context))))
(t/testing "the queue is a routing detail of the dispatcher, not context"
(t/is (not (contains? context :queue))))
(t/testing "nor is the tenant that owns the queue"
(t/is (not (contains? context :tenant))))
(t/testing "the retry keys belong to the runner, not to a handler"
(t/is (not (contains? context :retry-num)))
(t/is (not (contains? context :max-retries)))
(t/is (not (contains? context :attempt))))
(t/testing "the context is a plain map, not the row"
(t/is (not= (set (keys (mk-row {}))) (set (keys context)))))))
(t/deftest make-context-normalizes-an-empty-label
(t/testing "the empty label is the submit default for no label"
(t/is (nil? (:label (jobs/make-context (mk-row {:label ""})))))
(t/is (nil? (:label (jobs/make-context (mk-row {:label nil}))))))
(t/testing "a real label is carried as is"
(t/is (= "my-label" (:label (jobs/make-context (mk-row {:label "my-label"})))))))
(t/deftest make-context-always-carries-the-resource-reference
(t/testing "the key exists even when there is no resource"
(let [context (jobs/make-context (mk-row {}))]
(t/is (contains? context :resource-id))
(t/is (nil? (:resource-id context)))))
(t/testing "a resource reference is carried as is"
(let [resource-id (uuid/next)]
(t/is (= resource-id
(:resource-id (jobs/make-context (mk-row {:resource-id resource-id}))))))))
(t/deftest make-context-carries-the-job-owner
(t/testing "the key exists even when the job has no owner"
(let [context (jobs/make-context (mk-row {}))]
(t/is (contains? context :profile-id))
(t/is (nil? (:profile-id context)))))
(t/testing "the owner of a user job is carried as is"
(let [profile-id (uuid/next)]
(t/is (= profile-id
(:profile-id (jobs/make-context (mk-row {:profile-id profile-id}))))))))
(t/deftest make-context-rejects-invalid-rows
(t/testing "a missing id is rejected"
(t/is (thrown? Exception (jobs/make-context (mk-row {:id nil})))))
(t/testing "a missing name is rejected"
(t/is (thrown? Exception (jobs/make-context (mk-row {:name nil})))))
(t/testing "a non uuid owner is rejected"
(t/is (thrown? Exception (jobs/make-context (mk-row {:profile-id "nope"})))))
(t/testing "a non uuid resource reference is rejected"
(t/is (thrown? Exception (jobs/make-context (mk-row {:resource-id "nope"}))))))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; IN-PROCESS INVOCATION
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
(def received-contexts (atom []))
(defn- capture-defs
"Job-defs whose handler records the context it receives."
[]
{:echo {::jobs/name :echo
::jobs/schema schema:echo-params
::jobs/handler (fn [context params]
(swap! received-contexts conj context)
params)
::jobs/decoder (sm/decoder schema:echo-params sm/json-transformer)
::jobs/validator (sm/validator schema:echo-params)}})
(t/deftest invoke-delivers-a-nil-context-without-a-row
(reset! received-contexts [])
(let [cfg (make-cfg (capture-defs))
params (make-params)]
(t/is (= params (jobs/invoke (assoc cfg ::jobs/name :echo
::jobs/params params))))
(t/is (= [nil] @received-contexts))))
(t/deftest invoke-delivers-an-explicit-context
(reset! received-contexts [])
(let [cfg (make-cfg (capture-defs))
params (make-params)
context (jobs/make-context (mk-row {}))]
(jobs/invoke (assoc cfg ::jobs/name :echo
::jobs/params params
::jobs/context context))
(t/is (= [context] @received-contexts))
(t/testing "an explicit context does not enable heartbeat by itself"
(t/is (= [] (get-progresss (:id context)))))))
(t/deftest invoke-validates-an-explicit-context
(let [cfg (make-cfg (capture-defs))]
(t/testing "a partial context is rejected"
(t/is (thrown? Exception
(jobs/invoke (assoc cfg
::jobs/name :echo
::jobs/params (make-params)
::jobs/context {:id (uuid/next)})))))
(t/testing "an extra key is rejected: the schema is closed"
(t/is (thrown? Exception
(jobs/invoke (assoc cfg
::jobs/name :echo
::jobs/params (make-params)
::jobs/context
(assoc (jobs/make-context (mk-row {}))
:params {}))))))))
(t/deftest invoke-job-id-is-independent-from-the-context
(reset! received-contexts [])
(let [cfg (make-cfg (capture-defs))
params (make-params)
job-id (jobs/submit cfg {::jobs/name :echo
::jobs/params params})]
(jobs/invoke (assoc cfg ::jobs/name :echo
::jobs/params params
::jobs/job-id job-id))
(t/is (= [nil] @received-contexts)
"a job-id without a context still delivers a nil context")
(t/testing "the job-id is what makes the durable writes reach the row"
(t/is (pos? (jobs/heartbeat cfg :job-id job-id
:progress {:stage :work})))
(t/is (= [{:stage :work}] (get-progresss job-id))))))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; LIFECYCLE EVENTS
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
(defn- mk-running
"A job already claimed by a worker, ready for a terminal transition."
([cfg] (mk-running cfg :echo))
([cfg name]
(let [job-id (jobs/submit cfg {::jobs/name name
::jobs/params (make-params)})]
(jobs/claim cfg job-id (:scheduled-at (jobs/get-job cfg job-id)))
job-id)))
(t/deftest lifecycle-writers-require-metrics
;; Every one of the six writers must check the metrics instance before
;; it writes. Their metric lands inside a `db/after-commit` callback,
;; and those swallow exceptions, so without the check a miswired cfg
;; persists the row and loses the count with nothing but a log line.
(let [cfg (make-cfg (get-job-defs))
bare (dissoc cfg ::mtx/metrics)
count-jobs (fn [] (:cnt (th/db-exec-one! ["SELECT count(*) AS cnt FROM job"])))
state (fn [id] (:status (jobs/get-job cfg id)))]
(t/testing "the guards exist at all"
;; They are `clojure.core/assert` forms, and those are elided when
;; *assert* is false at compile time. Without the flag the checks
;; are not in the compiled code and every assertion below would
;; fail for a reason that has nothing to do with the writers.
(t/is (true? *assert*)
"run the suite with PENPOT_FLAGS=\"enable-backend-asserts\""))
(t/testing "submit never reaches the insert, and writes no row"
(let [before (count-jobs)]
(t/is (thrown-with-msg? Exception #"missing ::mtx/metrics"
(jobs/submit bare {::jobs/name :echo
::jobs/params (make-params)})))
(t/is (= before (count-jobs)))))
(t/testing "claim leaves the pending job pending"
(let [id (jobs/submit cfg {::jobs/name :echo
::jobs/params (make-params)})]
(t/is (thrown-with-msg? Exception #"missing ::mtx/metrics"
(jobs/claim bare id (:scheduled-at (jobs/get-job cfg id)))))
(t/is (= "new" (state id)))
(t/is (= [] (get-kinds id)))))
(t/testing "retry-job leaves the running job alone"
(let [id (mk-running cfg)]
(t/is (thrown-with-msg? Exception #"missing ::mtx/metrics"
(jobs/retry-job bare id 1 (ct/in-past {:minutes 1})
test-error :backoff)))
(t/is (= "running" (state id)))
(t/is (= ["start"] (get-kinds id)))))
(t/testing "complete leaves the running job alone"
(let [id (mk-running cfg)]
(t/is (thrown-with-msg? Exception #"missing ::mtx/metrics"
(jobs/complete bare :job-id id :result {:v 1})))
(t/is (= "running" (state id)))
(t/is (= ["start"] (get-kinds id)))))
(t/testing "fail leaves the running job alone"
(let [id (mk-running cfg)]
(t/is (thrown-with-msg? Exception #"missing ::mtx/metrics"
(jobs/fail bare id test-error)))
(t/is (= "running" (state id)))
(t/is (= ["start"] (get-kinds id)))))
(t/testing "cancel leaves the pending job pending"
(let [id (jobs/submit cfg {::jobs/name :echo
::jobs/params (make-params)})]
(t/is (thrown-with-msg? Exception #"missing ::mtx/metrics"
(jobs/cancel bare id)))
(t/is (= "new" (state id)))
(t/is (= [] (get-kinds id)))))))
(t/deftest lifecycle-writes-own-their-transaction
(let [cfg (make-cfg (get-job-defs))
;; a lifecycle write reports on work that already happened, so it
;; never joins the transaction of whoever called it. The row it
;; needs is committed beforehand: an independent connection
;; cannot see the uncommitted work of the transaction around it.
rolled-back
(fn [f]
(db/tx-run! (assoc cfg ::db/rollback true)
(fn [{:keys [::db/conn]}]
(f (assoc cfg ::db/conn conn)))))
state (fn [job-id]
{:status (:status (jobs/get-job cfg job-id))
:kinds (get-kinds job-id)})]
(t/testing "claim starts the job and stores the start event"
(let [job-id (jobs/submit cfg {::jobs/name :echo
::jobs/params (make-params)})
at (:scheduled-at (jobs/get-job cfg job-id))]
(t/is (= 1 (rolled-back #(jobs/claim % job-id at))))
(t/is (= {:status "running" :kinds ["start"]} (state job-id)))))
(t/testing "retry-job schedules the next attempt and stores the retry event"
(let [job-id (mk-running cfg)]
(t/is (= 1 (rolled-back #(jobs/retry-job % job-id 1 (ct/now) test-error :backoff))))
(t/is (= {:status "retry" :kinds ["start" "retry"]} (state job-id)))))
(t/testing "complete stores the end event with its result"
(let [job-id (mk-running cfg)]
(t/is (= 1 (rolled-back #(jobs/complete % :job-id job-id :result {:v 1}))))
(t/is (= {:status "completed" :kinds ["start" "end"]} (state job-id)))))
(t/testing "fail stores the end event with its error"
(let [job-id (mk-running cfg)]
(t/is (= 1 (rolled-back #(jobs/fail % job-id test-error))))
(t/is (= {:status "failed" :kinds ["start" "end"]} (state job-id)))))
(t/testing "cancel stores the end event"
(let [job-id (jobs/submit cfg {::jobs/name :echo
::jobs/params (make-params)})]
(t/is (= 1 (rolled-back #(jobs/cancel % job-id))))
(t/is (= {:status "cancelled" :kinds ["end"]} (state job-id)))))))
(t/deftest claim-records-a-start-event-with-the-attempt
(let [cfg (make-cfg (get-job-defs))]
(t/testing "the first attempt is 1"
(let [job-id (jobs/submit cfg {::jobs/name :echo
::jobs/params (make-params)})]
(jobs/claim cfg job-id (:scheduled-at (jobs/get-job cfg job-id)))
(t/is (= [{:kind "start" :payload {:attempt 1}}] (get-events job-id)))))
(t/testing "a retried job claims attempt retry-num + 1"
(let [job-id (jobs/submit cfg {::jobs/name :echo
::jobs/params (make-params)})]
(th/db-update! :job {:retry-num 2} {:id job-id})
(jobs/claim cfg job-id (:scheduled-at (jobs/get-job cfg job-id)))
(t/is (= [{:kind "start" :payload {:attempt 3}}] (get-events job-id)))))))
(t/deftest claim-that-affects-no-row-writes-no-event
(let [cfg (make-cfg (get-job-defs))
job-id (mk-running cfg)]
(t/is (= 0 (jobs/claim cfg job-id (:scheduled-at (jobs/get-job cfg job-id)))))
(t/testing "only the first claim left a start event"
(t/is (= ["start"] (get-kinds job-id))))))
(t/deftest retry-job-records-a-retry-event-per-reason
(let [cfg (make-cfg (get-job-defs))
job-id (mk-running cfg)]
(t/testing "a backoff increments the counter and reports the new attempt"
(jobs/retry-job cfg job-id 1 (ct/now) test-error :backoff)
(t/is (= {:attempt 2 :reason "backoff"}
(:payload (last (get-events job-id))))))
(t/testing "a noop keeps the counter and re-reports the same attempt"
(jobs/retry-job cfg job-id 1 (ct/now) test-error :noop)
(t/is (= {:attempt 2 :reason "noop"}
(:payload (last (get-events job-id))))))
(t/testing "the counter on the row is the one we asked for"
(t/is (= 1 (:retry-num (jobs/get-job cfg job-id)))))
(t/testing "a terminal job gets no retry and no event"
(th/db-update! :job {:status "failed"} {:id job-id})
(t/is (= 0 (jobs/retry-job cfg job-id 2 (ct/now) test-error :backoff)))
(t/is (= ["start" "retry" "retry"] (get-kinds job-id))))))
(t/deftest complete-records-an-end-event
(let [cfg (make-cfg (get-job-defs))
job-id (mk-running cfg)]
(t/is (= 1 (jobs/complete cfg :job-id job-id :result {:value 1})))
(t/is (= ["start" "end"] (get-kinds job-id)))
(t/is (= ["completed"] (get-outcomes job-id)))
(t/testing "a terminal job gets no second end event"
(t/is (= 0 (jobs/complete cfg :job-id job-id :result {:value 2})))
(t/is (= ["completed"] (get-outcomes job-id))))))
(t/deftest fail-records-an-end-event-and-a-structured-error
(let [cfg (make-cfg (get-job-defs))
job-id (mk-running cfg)]
(t/is (= 1 (jobs/fail cfg job-id test-error)))
(t/is (= ["failed"] (get-outcomes job-id)))
(t/testing "the error decodes back to the map we passed"
(t/is (= test-error (jobs/decode-job-error (:error (jobs/get-job cfg job-id))))))
(t/testing "the event carries no error, only the outcome"
(t/is (= {:outcome "failed"} (:payload (last (get-events job-id))))))
(t/testing "a terminal job gets no second end event"
(t/is (= 0 (jobs/fail cfg job-id test-error)))
(t/is (= ["failed"] (get-outcomes job-id))))))
(t/deftest cancel-records-an-end-event
(let [cfg (make-cfg (get-job-defs))
job-id (jobs/submit cfg {::jobs/name :echo
::jobs/params (make-params)})]
(t/is (= 1 (jobs/cancel cfg job-id)))
(t/is (= ["cancelled"] (get-outcomes job-id)))
(t/testing "a terminal job gets no second end event"
(t/is (= 0 (jobs/cancel cfg job-id)))
(t/is (= ["cancelled"] (get-outcomes job-id))))))
(t/deftest get-job-decodes-the-result-with-the-schema-of-the-def
"A result with a declared schema comes back typed. Without it, the plain
JSON of the column keywordizes every key and leaves every uuid as text;
without the substrate compiling the decoder from that schema, the key of
the uuid map comes back as a keyword and this test fails."
(let [cfg (make-cfg (get-job-defs))
job-id (mk-running cfg :typed)
file-id (uuid/next)
lib-id (uuid/next)]
(t/is (= 1 (jobs/complete cfg :job-id job-id
:result {:file-ids [lib-id]
:resolution {file-id {:id file-id}}})))
(let [result (:result (jobs/get-job cfg job-id))]
(t/testing "the uuid keyed map keeps the type of its key and values"
(t/is (= [file-id] (keys (:resolution result))))
(t/is (uuid? (:id (get (:resolution result) file-id)))))
(t/testing "the vector of uuids keeps the type of its elements"
(t/is (= [lib-id] (:file-ids result)))))))
(t/deftest get-job-leaves-a-result-without-a-schema-untouched
(let [cfg (make-cfg (get-job-defs))
job-id (mk-running cfg)
file-id (uuid/next)]
(t/is (= 1 (jobs/complete cfg :job-id job-id
:result {:resolution {file-id {:id file-id}}})))
(t/testing "the raw JSON shape survives: keywordized keys, uuid as text"
(let [result (:result (jobs/get-job cfg job-id))
key (keyword (str file-id))]
(t/is (= [key] (keys (:resolution result))))
(t/is (= (str file-id) (get-in result [:resolution key :id])))))))
(t/deftest get-job-reads-a-row-whose-job-def-is-gone
(let [job-id (mk-running (make-cfg (get-job-defs)))
;; a registry that no longer knows the kind of the row (a retired
;; job-def): reading its result must not raise
row (jobs/get-job (make-cfg {}) job-id)]
(t/is (some? row))
(t/is (= "echo" (:name row)))))
(t/deftest complete-refuses-a-result-that-does-not-conform
"The schema a job-def declares is checked before the result is stored: the
row keeps no result and the job is not completed."
(let [cfg (make-cfg (get-job-defs))
job-id (mk-running cfg :typed)]
(t/is (thrown-with-msg? Exception #"invalid result of a job"
(jobs/complete cfg :job-id job-id
:result {:resolution "not-a-map"})))
(t/testing "the row was not completed and holds no result"
(let [row (jobs/get-job cfg job-id)]
(t/is (= "running" (:status row)))
(t/is (nil? (:result row)))))
(t/testing "a job that is no longer completable is skipped without checking"
(th/db-update! :job {:status "aborted"} {:id job-id})
(t/is (zero? (jobs/complete cfg :job-id job-id
:result {:resolution "not-a-map"}))))))
(t/deftest a-stored-result-that-does-not-conform-is-read-decode-only
"A row written by another version may hold a result the schema of today no
longer accepts: reading it decodes and never raises."
(let [cfg (make-cfg (get-job-defs))
job-id (mk-running cfg :typed)]
(th/db-update! :job
{:status "completed"
:result (db/json {:resolution "not-a-map"})}
{:id job-id})
(let [row (jobs/get-job cfg job-id)]
(t/is (= "completed" (:status row)))
(t/is (= {:resolution "not-a-map"} (:result row))))))
(t/deftest decode-result-accepts-a-pgobject
(let [job-def (jobs/get-job-def (get (make-cfg (get-job-defs)) ::jobs/defs) :typed)
file-id (uuid/next)]
(t/testing "a raw column value is decoded with the schema of the def"
(let [decoded (jobs/decode-result job-def
(db/json {:file-ids [file-id]
:resolution {file-id {:id file-id}}}))]
(t/is (= [file-id] (:file-ids decoded)))
(t/is (= [file-id] (keys (:resolution decoded))))))
(t/testing "no result answers nil"
(t/is (nil? (jobs/decode-result job-def nil))))))
(t/deftest aborted-job-is-terminal-for-all-writers
"An aborted job (system-side terminal set by the orphan sweep) cannot be
moved by any lifecycle writer: first-terminal-wins, with no event and no
second transition."
(let [cfg (make-cfg (get-job-defs))
job-id (mk-running cfg)
at (:scheduled-at (jobs/get-job cfg job-id))]
(th/db-update! :job {:status "aborted"
:error (db/json jobs/orphan-error)}
{:id job-id})
(t/is (zero? (jobs/claim cfg job-id at)))
(t/is (zero? (jobs/retry-job cfg job-id 1 (ct/now) test-error :backoff)))
(t/is (zero? (jobs/complete cfg :job-id job-id :result {:v 1})))
(t/is (zero? (jobs/fail cfg job-id test-error)))
(t/is (zero? (jobs/cancel cfg job-id)))
(t/is (= "aborted" (:status (jobs/get-job cfg job-id))))
(t/is (= ["start"] (get-kinds job-id))
"only the claim start event, no end event from any writer")))
(defn- with-failing-event
"Run `f` with the event insert forced to fail, the way a database that
rejects the row would. Everything the transition writes must go back
with the transaction that carried it."
[f]
(with-redefs [jobs/insert-event (fn [& _]
(throw (ex-info "event insert failed" {})))]
(f)))
(defn- job-columns
"Every column a transition can move, decoded, so a before/after
comparison is a real one."
[cfg job-id]
(let [row (jobs/get-job cfg job-id)]
(cond-> (select-keys row [:status :modified-at :completed-at :retry-num
:scheduled-at :resource-id :result :error])
(:error row) (update :error db/decode-json-pgobject)
(:result row) (update :result db/decode-json-pgobject))))
(t/deftest a-failing-event-write-rolls-back-every-writer
(let [cfg (make-cfg (get-job-defs))
unchanged
(fn [job-id before]
(t/is (= before (job-columns cfg job-id))
"no column the transition touches survived")
(t/is (not-any? #(= "end" %) (get-kinds job-id))
"and no event of the failed transition was left behind"))]
(t/testing "claim"
(let [job-id (jobs/submit cfg {::jobs/name :echo
::jobs/params (make-params)})
at (:scheduled-at (jobs/get-job cfg job-id))
before (job-columns cfg job-id)]
(with-failing-event #(t/is (thrown? Exception (jobs/claim cfg job-id at))))
(unchanged job-id before)
(t/is (= [] (get-kinds job-id)))))
(t/testing "retry-job"
(let [job-id (mk-running cfg)
before (job-columns cfg job-id)]
(with-failing-event
#(t/is (thrown? Exception
(jobs/retry-job cfg job-id 1 (ct/in-past {:minutes 1})
test-error :backoff))))
(unchanged job-id before)
(t/is (= ["start"] (get-kinds job-id)))))
(t/testing "complete"
(let [job-id (mk-running cfg)
before (job-columns cfg job-id)]
(with-failing-event
#(t/is (thrown? Exception
(jobs/complete cfg :job-id job-id :result {:v 1}))))
(unchanged job-id before)
(t/is (= ["start"] (get-kinds job-id)))))
(t/testing "fail"
(let [job-id (mk-running cfg)
before (job-columns cfg job-id)]
(with-failing-event #(t/is (thrown? Exception (jobs/fail cfg job-id test-error))))
(unchanged job-id before)
(t/is (= ["start"] (get-kinds job-id)))))
(t/testing "cancel"
(let [job-id (jobs/submit cfg {::jobs/name :echo
::jobs/params (make-params)})
before (job-columns cfg job-id)]
(with-failing-event #(t/is (thrown? Exception (jobs/cancel cfg job-id))))
(unchanged job-id before)
(t/is (= [] (get-kinds job-id)))))
(t/testing "a resource a failed completion associated is undone with it"
(let [job-id (mk-running cfg)
resource-id (uuid/next)]
(th/db-insert! :storage-object {:id resource-id :backend "test"})
(with-failing-event
#(t/is (thrown? Exception
(jobs/complete cfg :job-id job-id :result {:v 1}
:resource-id resource-id))))
(let [row (jobs/get-job cfg job-id)]
(t/is (nil? (:resource-id row)))
(t/is (= "running" (:status row))))))))
(t/deftest insert-event-requires-a-transaction
;; The event insert is only valid inside the transaction that owns the
;; job row: outside one the assert fires instead of writing. Like every
;; assert-based guard, this only holds with the backend asserts flag.
(t/is (true? *assert*)
"run the suite with PENPOT_FLAGS=\"enable-backend-asserts\"")
(let [cfg (make-cfg (get-job-defs))
job-id (mk-running cfg)]
(t/testing "a pool cfg with no transaction raises"
(t/is (thrown? AssertionError
(@#'jobs/insert-event cfg job-id "progress" {:stage :half}))))
(t/testing "a bare connection with no transaction raises too"
(db/run! cfg (fn [conn-cfg]
(t/is (thrown? AssertionError
(@#'jobs/insert-event conn-cfg job-id "progress"
{:stage :half}))))))))
(t/deftest complete-associates-a-resource-id-once
(let [cfg (make-cfg (get-job-defs))
resource-id (uuid/next)]
(th/db-insert! :storage-object {:id resource-id :backend "test"})
(t/testing "a job without resource gets one at completion"
(let [job-id (mk-running cfg)]
(t/is (= 1 (jobs/complete cfg :job-id job-id
:result {:v 1}
:resource-id resource-id)))
(t/is (= resource-id (:resource-id (jobs/get-job cfg job-id))))))
(t/testing "completing without a resource leaves the column untouched"
(let [job-id (mk-running cfg)]
(t/is (= 1 (jobs/complete cfg :job-id job-id :result {:v 1})))
(t/is (nil? (:resource-id (jobs/get-job cfg job-id))))))
(t/testing "a job that already has a resource refuses a second one"
(let [job-id (mk-running cfg)]
;; a job that already carries a resource, as submit left it
(th/db-update! :job {:resource-id resource-id} {:id job-id})
(t/is (thrown-with-msg? Exception #"already has a resource"
(jobs/complete cfg
:job-id job-id
:result {:v 1}
:resource-id (uuid/next))))
(t/testing "and the job is still running"
(t/is (= "running" (:status (jobs/get-job cfg job-id)))))))
(t/testing "a resource that is not a uuid is rejected before any write"
(let [job-id (mk-running cfg)]
(t/is (thrown-with-msg? Exception #"must be a uuid"
(jobs/complete cfg
:job-id job-id
:result {:v 1}
:resource-id "nope")))
(t/is (= "running" (:status (jobs/get-job cfg job-id))))))))
(t/deftest job-error-schema-is-shared-and-validated
(t/testing "type and code are keywords so they decode back from JSON"
(t/is (= {:type :internal :code :orphan :hint "gone"}
(jobs/check-job-error {:type :internal :code :orphan :hint "gone"}))))
(t/testing "a missing key is rejected"
(t/is (thrown? Exception (jobs/check-job-error {:type :internal :code :orphan})))
(t/is (thrown? Exception (jobs/check-job-error {:type :internal
:code "orphan"
:hint "gone"}))))
(t/testing "extra details are allowed"
(t/is (= {:type :internal :code :orphan :hint "gone" :attempt 3}
(jobs/check-job-error {:type :internal
:code :orphan
:hint "gone"
:attempt 3}))))
(t/testing "fail validates before writing"
(let [cfg (make-cfg (get-job-defs))
job-id (mk-running cfg)]
(t/is (thrown? Exception (jobs/fail cfg job-id {:code :orphan})))
(t/is (= "running" (:status (jobs/get-job cfg job-id))))
(t/is (= ["start"] (get-kinds job-id)))))
(t/testing "the orphan error is a valid job error"
(t/is (= jobs/orphan-error (jobs/check-job-error jobs/orphan-error)))))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; CONTENTION
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; The safety of the writers rests on two things: a conditional UPDATE
;; that only the first caller can win, and a row lock the loser waits
;; on. Both are only meaningful when two callers arrive at once, so the
;; tests below put real threads in front of them.
(defn- race
"Run every thunk on a thread of its own and return what each returned,
in the order they were given. They all wait on the same gate, so they
contend for whatever they share instead of running one after another."
[& thunks]
(let [gate (promise)
results (mapv (fn [f]
(future
@gate
(f)))
thunks)]
(deliver gate true)
(mapv deref results)))
(t/deftest the-first-terminal-transition-wins
(let [cfg (make-cfg (get-job-defs))
job-id (mk-running cfg)]
(t/testing "a completion and a failure of the same job, at the same time"
(let [[completed failed] (race #(jobs/complete cfg :job-id job-id :result {:v 1})
#(jobs/fail cfg job-id test-error))]
(t/is (= #{0 1} (set [completed failed]))
"one transition is applied and the other affects no row"))
(t/testing "and the durable history agrees with the row that won"
(let [kinds (get-kinds job-id)
outcomes (get-outcomes job-id)
row (jobs/get-job cfg job-id)]
(t/is (= ["start" "end"] kinds)
"exactly one end event, written by whoever won")
(t/is (= 1 (count outcomes)))
(t/is (= (first outcomes) (:status row))
"the stored outcome is the status of the row"))))))
(t/deftest the-first-claim-wins
(let [cfg (make-cfg (get-job-defs))
job-id (jobs/submit cfg {::jobs/name :echo
::jobs/params (make-params)})
at (:scheduled-at (jobs/get-job cfg job-id))]
(t/testing "two workers claim the same advertised job at the same time"
(let [[a b] (race #(jobs/claim cfg job-id at)
#(jobs/claim cfg job-id at))]
(t/is (= [0 1] (sort [a b])))
(t/is (= "running" (:status (jobs/get-job cfg job-id))))
(t/is (= ["start"] (get-kinds job-id))
"a job is started once however many workers ask for it")))))
(t/deftest a-report-never-lands-on-a-terminal-job
(let [cfg (make-cfg (get-job-defs))
job-id (mk-running cfg)]
(t/testing "a progress report and a completion of the same job, at once"
;; the report either lands before the end or loses the race and gets
;; the interrupt: a beat on a job that is no longer active raises
(let [report (fn []
(if (some? (raised #(jobs/heartbeat cfg :job-id job-id
:progress {:stage :work}
::jobs/force? true)))
:interrupted
:reported))
[outcome completed] (race report #(jobs/complete cfg :job-id job-id :result {:v 1}))]
(t/is (contains? #{:interrupted :reported} outcome)
"the report lands or gets the interrupt, never anything else")
(t/is (= 1 completed) "the completion always wins, it owns the row")
(t/testing "the report either lost the race or was stored before the end"
(let [kinds (get-kinds job-id)]
(t/is (= "start" (first kinds)))
(t/is (= "end" (last kinds))
"the end is always the last event, so no report came after it")
(t/is (<= 2 (count kinds) 3)
"at most the one report that made it in before the end"))))
(t/testing "a job that ended is not resurrected by a late report"
(let [before (get-kinds job-id)
cause (raised #(jobs/heartbeat cfg :job-id job-id
:progress {:stage :work}
::jobs/force? true))]
(t/is (= :interrupt (th/ex-type cause)))
(t/is (= before (get-kinds job-id)) "the late report stored nothing"))))))
(t/deftest progress-event-reaches-the-cfg-of-its-own-callback
"The event helper needs the msgbus and the metrics from the cfg, and it
must find them on the map that `db/tx-run!` hands to the callback: the
writers pass the whole cfg, not a rebuilt one. A cfg missing either
dependency cannot produce a published event, so the publication is
what proves the cfg arrived intact."
(let [cfg (make-cfg (get-job-defs))
profile-id (:id (th/create-profile* 1))
messages (atom [])
job-cfg (assoc cfg ::mbus/msgbus (fake-msgbus messages))
job-id (jobs/submit job-cfg {::jobs/name :echo
::jobs/params (make-params)
::jobs/profile-id profile-id})]
(t/is (pos? (jobs/heartbeat job-cfg :job-id job-id
:progress {:stage :work})))
(t/testing "the event was stored"
(t/is (= [{:stage :work}] (get-progresss job-id))))
(t/testing "and the msgbus on that cfg was reachable from the callback"
(t/is (= 1 (count @messages)))
(t/is (= "progress" (:kind (:message (first @messages))))))))
Reference in New Issue View Git Blame Copy Permalink