✨ Complete jobs with the multipart result and skip status

`complete-job` accepts a second form: the worker sends the artifact
itself as a multipart body, the server stores it in the `job-resource`
bucket owned by the job profile, and the completion coalesces
`resource_id` and the result descriptor `{resource-uri, filename,
mtype, size}` in its own transaction (first terminal state wins; a
lost race releases the artifact it stored, so nothing lingers for the
storage GC). The checks run before any object is written: a lost race
against a cancel never stores bytes, the artifact type must be one the
export produces (new `cm/export-artifact-types` allowlist) and the size
must fit the new `:exporter-max-result-size` config (default 350 MiB)
below the HTTP multipart ceiling.

`complete-job` and `fail-job` take an optional `session-id`, deleted
best-effort after the commit, and every write method answers `:skip`
with the `:status` of the row it lost against, read fresh after the
failed write, so a worker tells a cancelled job apart from one it was
too late for.

AI-assisted-by: glm-5.3-flash
This commit is contained in:
Andrey Antukh 2026-10-07 07:42:48 +00:00
parent 53008ea794
commit 8328974f2e
4 changed files with 356 additions and 39 deletions

View File

@ -82,6 +82,7 @@
:jobs-user-ttl (ct/duration {:days 7})
:jobs-request-timeout (ct/duration {:minutes 2})
:job-session-ttl (ct/duration {:hours 1})
:exporter-max-result-size (* 1024 1024 350) ; 350MiB
:media-max-file-size (* 1024 1024 30) ; 30MiB
:font-max-file-size (* 1024 1024 30) ; 30MiB
@ -203,6 +204,7 @@
[:jobs-user-ttl {:optional true} ::ct/duration]
[:jobs-request-timeout {:optional true} ::ct/duration]
[:job-session-ttl {:optional true} ::ct/duration]
[:exporter-max-result-size {:optional true} ::sm/int]
[:telemetry-enabled {:optional true} ::sm/boolean]
[:default-blob-version {:optional true} ::sm/int]
[:allow-demo-users {:optional true} ::sm/boolean]
@ -501,6 +503,16 @@
(or (c/get config :job-session-ttl)
(ct/duration {:hours 1})))
(defn get-exporter-max-result-size
"How big the artifact a worker sends with `complete-job` may be. The
ceiling is the same the multipart body already has at the HTTP server
(`:http/max-multipart-body-size`); lowering it rejects a result
before any object is stored. Raising it past the server cap is an
operational change of both this and nginx `client_max_body_size`."
[]
(or (c/get config :exporter-max-result-size)
(* 1024 1024 350)))
(defn get
"A configuration getter. Helps code be more testable."
([key]

View File

@ -14,18 +14,27 @@
do not require a profile. Actual authentication is enforced by the
management route resolver (resolve-management-methods in app.rpc),
which mandates a valid shared-key before dispatching to these methods.
If the route resolver changes, these methods must be updated accordingly."
If the route resolver changes, these methods must be updated accordingly.
Every write method answers `{:action :run}` when it landed and
`{:action :skip}` when it did not, with the state of the row it found
(read fresh after the failed write) in `:status`: the worker tells a
cancelled job apart from one it was too late for, without guessing."
(:require
[app.common.exceptions :as ex]
[app.common.media :as cm]
[app.common.schema :as sm]
[app.common.time :as ct]
[app.config :as cf]
[app.db :as db]
[app.http.session :as session]
[app.jobs :as jobs]
[app.jobs.storage :as js]
[app.media.validation :as media.v]
[app.rpc :as-alias rpc]
[app.rpc.doc :as doc]
[app.setup :as-alias setup]
[app.storage :as sto]
[app.util.services :as sv]))
;; ---- RPC METHOD: CLAIM-JOB
@ -40,7 +49,24 @@
[:action [:enum :run :skip]]
[:name {:optional true} ::sm/text]
;; Opaque per-job params blob; intentionally untyped.
[:params {:optional true} :any]])
[:params {:optional true} :any]
;; On `:skip`, the state of the row the write lost against.
[:status {:optional true} ::sm/text]])
(defn- row-status
"The state of the row after a write that touched nothing: one fresh
read, never the row the method already saw, so the answer names the
state the race produced."
[cfg job-id]
(:status (jobs/get-job cfg job-id)))
(defn- skip-with-status
"The answer of a landed-nowhere write: one fresh read of the row
after the failure, so the worker sees the state the race produced."
[cfg job-id]
(if-let [status (row-status cfg job-id)]
{:action :skip :status status}
{:action :skip}))
(sv/defmethod ::claim-job
{::doc/added "2.20"
@ -58,7 +84,7 @@
{:action :run
:name (:name row)
:params (:params row)}
{:action :skip})))
(skip-with-status cfg job-id))))
;; ---- RPC METHOD: REPORT-JOB-PROGRESS
@ -72,7 +98,8 @@
(def ^:private schema:report-job-progress-result
[:map {:title "report-job-progress-result"}
[:action [:enum :run :skip]]])
[:action [:enum :run :skip]]
[:status {:optional true} ::sm/text]])
(sv/defmethod ::report-job-progress
{::doc/added "2.20"
@ -94,44 +121,169 @@
(cond
(ex/exception? result)
(if (= :interrupt (:type (ex-data result)))
{:action :skip}
(skip-with-status cfg job-id)
(throw result))
(pos? result)
{:action :run}
:else
{:action :skip})))
(skip-with-status cfg job-id))))
;; ---- RPC METHOD: COMPLETE-JOB
(def ^:private schema:complete-job-params
[:map {:title "complete-job-params"}
[:job-id ::sm/uuid]
;; Per-job result: any JSON-encodable value, nil when the job returns
;; nothing. Nested maps and vectors are allowed; unserializable
;; values are dropped to nil by jobs/encode-result with a warning.
[:result [:maybe :any]]
;; Storage object the worker produced for this job. A resource is only
;; set, never replaced: a job that already has one is rejected.
[:resource-id {:optional true} ::sm/uuid]])
[:and
[:map {:title "complete-job-params"}
[:job-id ::sm/uuid]
;; Per-job result: any JSON-encodable value, nil when the job returns
;; nothing. Nested maps and vectors are allowed; unserializable
;; values are dropped to nil by jobs/encode-result with a warning.
[:result {:optional true} [:maybe :any]]
;; Storage object the worker produced for this job. A resource is only
;; set, never replaced: a job that already has one is rejected.
[:resource-id {:optional true} ::sm/uuid]
;; The multipart form: the worker sends the artifact itself, the
;; server stores it. `filename` and `mtype` name what the user will
;; download; they say nothing about the bytes (the `content` field
;; carries its own type).
[:content {:optional true} media.v/schema:upload]
[:filename {:optional true} ::sm/text]
[:mtype {:optional true} ::sm/text]
;; The render session the worker used, closed once the job settles:
;; complete is terminal, the session has no reader after it.
[:session-id {:optional true} ::sm/uuid]]
[:fn {:error/message "one of :content, :result or :resource-id is required"}
(fn [{:keys [content resource-id] :as params}]
(boolean (or (some? content)
(contains? params :result)
(some? resource-id))))]])
(def ^:private schema:complete-job-result
[:map {:title "complete-job-result"}
[:action [:enum :run :skip]]])
[:action [:enum :run :skip]]
[:status {:optional true} ::sm/text]])
(defn- close-job-session!
"The session a worker rendered with, deleted once the job settles.
Best-effort: the settle already decided the outcome, and a session
that cannot be deleted now is swept later by the sessions idle GC."
[cfg session-id]
(when (uuid? session-id)
(ex/ignoring (session/delete-session (::session/manager cfg) session-id))))
(defn- check-artifact!
"The bytes a worker offers as the result: the type must be one the
export produces and the size under the configured cap. Both checks
refuse to *use* the bytes, not to receive them (the multipart parser
already wrote them to a server tempfile); refusing early costs
nothing, storing early costs a GC cycle."
[content]
(media.v/validate-media-type! content cm/export-artifact-types)
(let [max-size (cf/get-exporter-max-result-size)]
(when (> (:size content) max-size)
(ex/raise :type :validation
:code :request-body-too-large
:hint "the result of the job is over the maximum size"
:size (:size content)
:max-size max-size)))
content)
(defn- completable-job
"The row a completion may land on: running (or retrying), whatever
the JSON form asks for."
[cfg job-id]
(let [job (jobs/get-job cfg job-id)]
(when (and (some? job)
(contains? #{"running" "retry"} (:status job)))
job)))
(defn- artifact-owner
"The job an artifact completion may land on: completable AND owned,
because the object needs an owner and a job without one owns nothing
(it may only settle without an artifact). Read before anything is
stored: a lost race with a cancel never turns the bytes of the worker
into an object."
[cfg job-id]
(let [job (completable-job cfg job-id)]
(when (and (some? job)
(uuid? (:profile-id job)))
job)))
(defn- complete-with-artifact
"The multipart form of a completion, on a job that is still
completable: the artifact is stored out of any transaction (storage
waits for no database), owned by the profile of the job, and the
completion itself writes the `resource_id` and the result descriptor
`{resource-uri filename mtype size}` in its own transaction, where
the first terminal state wins. A completion that lost the race
releases the artifact it stored: no object lingers for the storage GC
to find later."
[cfg job content filename mtype session-id]
(let [resource (js/put-resource cfg
(:profile-id job)
{:content (sto/content (:path content))
:filename filename
:mtype mtype})]
(if (pos? (jobs/complete cfg
:job-id (:id job)
:resource-id (:resource-id resource)
:result {:resource-uri (:resource-uri resource)
:filename filename
:mtype mtype
:size (:size resource)}))
(do
(close-job-session! cfg session-id)
{:action :run})
(do
(js/release-resource cfg (:resource-id resource))
(skip-with-status cfg (:id job))))))
(sv/defmethod ::complete-job
"Close a running job as completed, in one of two forms.
The JSON form carries the plain `:result` the job produced and an
optional storage object (`:resource-id`) the worker wrote itself: the
internal binfile path.
The multipart form carries the artifact itself (`:content`, with
`:filename` and `:mtype` naming the download): the server stores it,
owned by the profile of the job, and coalesces the resource into the
same completion. The checks (completable, type, size) run before any
object is written: a lost race against a cancel never stores bytes.
With `:session-id`, the render session the worker minted through
`create-job-session` is deleted after the completion commits.
A completion of a job that already ended answers `{:action :skip,
:status ...}` and changes nothing."
{::doc/added "2.20"
::sm/params schema:complete-job-params
::sm/result schema:complete-job-result
::rpc/auth false} ;; shared-key enforced by route resolver
[cfg {:keys [job-id result resource-id]}]
(if (pos? (jobs/complete cfg
:job-id job-id
:result result
:resource-id resource-id))
{:action :run}
{:action :skip}))
[cfg {:keys [job-id result resource-id content filename mtype session-id]}]
(cond
;; the multipart form stores the artifact itself: only for a job
;; that can still complete and has an owner, because the object
;; needs one
(some? content)
(if-let [job (artifact-owner cfg job-id)]
(do
(check-artifact! content)
(complete-with-artifact cfg job content filename mtype session-id))
(skip-with-status cfg job-id))
;; the JSON form: nothing to store here, the resource (if any)
;; already lives in storage
:else
(if (pos? (jobs/complete cfg
:job-id job-id
:result result
:resource-id resource-id))
(do
(close-job-session! cfg session-id)
{:action :run})
(skip-with-status cfg job-id))))
;; ---- RPC METHOD: FAIL-JOB
@ -140,21 +292,27 @@
[:job-id ::sm/uuid]
;; Rich error report: type/code/hint are required, the map stays
;; open to worker-defined details (see jobs/fail).
[:error jobs/schema:job-error]])
[:error jobs/schema:job-error]
;; The render session the worker used, closed once the job settles:
;; fail is terminal too.
[:session-id {:optional true} ::sm/uuid]])
(def ^:private schema:fail-job-result
[:map {:title "fail-job-result"}
[:action [:enum :run :skip]]])
[:action [:enum :run :skip]]
[:status {:optional true} ::sm/text]])
(sv/defmethod ::fail-job
{::doc/added "2.20"
::sm/params schema:fail-job-params
::sm/result schema:fail-job-result
::rpc/auth false} ;; shared-key enforced by route resolver
[cfg {:keys [job-id error]}]
[cfg {:keys [job-id error session-id]}]
(if (pos? (jobs/fail cfg job-id error))
{:action :run}
{:action :skip}))
(do
(close-job-session! cfg session-id)
{:action :run})
(skip-with-status cfg job-id)))
;; ---- RPC METHOD: CREATE-JOB-SESSION

View File

@ -15,8 +15,12 @@
[app.jobs :as jobs]
[app.metrics :as-alias mtx]
[app.setup :as-alias setup]
[app.storage.tmp :as tmp]
[backend-tests.helpers :as th]
[clojure.test :as t]
[cuerdas.core :as str]
[datoteka.fs :as fs]
[datoteka.io :as io]
[mockery.core :refer [with-mocks]]))
(t/use-fixtures :once th/state-init)
@ -59,8 +63,8 @@
(defn- get-row
[id]
(let [row (th/db-get :job {:id id}
:status :result :error
:started-at :completed-at :modified-at)]
:status :result :error :resource-id
:started-at :completed-at :modified-at :name)]
(reduce (fn [row key]
(update row key #(cond-> % (db/pgobject? %) db/decode-json-pgobject)))
row
@ -124,20 +128,22 @@
out (mgmt :claim-job {:job-id job-id
:scheduled-at (ct/in-future {:minutes 5})})]
(t/is (nil? (:error out)))
(t/is (= {:action :skip} (:result out)))
(t/is (= {:action :skip :status "new"} (:result out)))
(t/is (= "new" (:status (get-row job-id))))))
(t/deftest claim-job-skips-terminal-and-cancelled-rows
;; the skip says the state of the row that won: a worker that carries
;; a stale payload sees cancelled/aborted, not a plain skip to retry on
(let [completed-id (mk-job {:status "completed"})
cancelled-id (mk-job {:status "cancelled"})
aborted-id (mk-job {:status "aborted"})]
(t/is (= {:action :skip}
(t/is (= {:action :skip :status "completed"}
(:result (mgmt :claim-job {:job-id completed-id
:scheduled-at (ct/now)}))))
(t/is (= {:action :skip}
(t/is (= {:action :skip :status "cancelled"}
(:result (mgmt :claim-job {:job-id cancelled-id
:scheduled-at (ct/now)}))))
(t/is (= {:action :skip}
(t/is (= {:action :skip :status "aborted"}
(:result (mgmt :claim-job {:job-id aborted-id
:scheduled-at (ct/now)}))))
(t/is (= "completed" (:status (get-row completed-id))))
@ -172,7 +178,7 @@
:progress {:stage :pages
:counters {:pages {:current 100 :total 100}}}})]
(t/is (nil? (:error out)))
(t/is (= {:action :skip} (:result out)))
(t/is (= {:action :skip :status "completed"} (:result out)))
(t/is (= [] (get-progresss job-id)))))
(t/deftest report-job-progress-persists-rapid-reports
@ -314,19 +320,19 @@
(t/testing "complete on a terminal row reports skip"
(let [out (mgmt :complete-job {:job-id done-id :result {:x 1}})]
(t/is (nil? (:error out)))
(t/is (= {:action :skip} (:result out)))))
(t/is (= {:action :skip :status "completed"} (:result out)))))
(t/testing "fail on a terminal row reports skip"
(let [out (mgmt :fail-job {:job-id done-id
:error {:type :internal
:code "processing-error"
:hint "bad image"}})]
(t/is (nil? (:error out)))
(t/is (= {:action :skip} (:result out)))))
(t/is (= {:action :skip :status "completed"} (:result out)))))
(t/testing "progress on a terminal row reports skip"
(let [out (mgmt :report-job-progress {:job-id done-id
:progress {:stage :pages :counters {:pages {:current 4 :total 10}}}})]
(t/is (nil? (:error out)))
(t/is (= {:action :skip} (:result out))))))
(t/is (= {:action :skip :status "completed"} (:result out))))))
(let [cfg (make-cfg)
job-id (mk-job {})
_ (jobs/claim cfg job-id (:scheduled-at (th/db-get :job {:id job-id} :id :scheduled-at)))]
@ -385,7 +391,7 @@
:result {:v 2}
:resource-id second})]
(t/is (nil? (:error out)))
(t/is (= {:action :skip} (:result out)))))
(t/is (= {:action :skip :status "completed"} (:result out)))))
(t/testing "the row keeps the resource it was given and no event is added"
(let [row (get-row job-id)]
@ -547,3 +553,132 @@
(let [out (mgmt :create-job-session {:job-id job-id})]
(t/is (= :not-found (th/ex-type (:error out))))
(t/is (= :profile-not-found (th/ex-code (:error out))))))))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; COMPLETE-JOB MULTIPART
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
(defn- artifact-tempfile
"An on-disk artifact, as the multipart parser would leave one for the
server: a path with the bytes of the produced export in it."
[content]
(let [path (fs/create-tempfile :dir tmp/default-tmp-dir
:prefix "test-artifact-")]
(with-open [out (io/output-stream path)]
(io/write* out (.getBytes ^String content "UTF-8")))
path))
(defn- multipart-complete
[job-id path & {:as extra}]
(mgmt :complete-job (merge {:job-id job-id
:content {:path path
:filename "result.zip"
:size (fs/size path)
:mtype "application/zip"}
:filename "result.zip"
:mtype "application/zip"}
extra)))
(t/deftest complete-job-multipart-stores-the-artifact-and-completes
(let [profile (mk-profile)
job-id (mk-job {:status "running"
:profile-id (:id profile)})
path (artifact-tempfile "penpot")
out (multipart-complete job-id path)
row (get-row job-id)
rid (:resource-id row)]
(t/is (nil? (:error out)))
(t/is (= {:action :run} (:result out)))
(t/is (= "completed" (:status row)))
(t/testing "the artifact lives in the job-resource bucket, owned by the profile"
(let [{:keys [metadata]} (first (th/db-exec! ["SELECT metadata
FROM storage_object WHERE id = ?" rid]))
metadata (cond-> metadata (db/pgobject? metadata) db/decode-transit-pgobject)]
(t/is (some? rid))
(t/is (= "job-resource" (:bucket metadata)))
(t/is (= (:id profile) (:profile-id metadata)))))
(t/testing "the result names the artifact and how to download it"
(let [result (:result row)]
(t/is (= "result.zip" (:filename result)))
(t/is (= "application/zip" (:mtype result)))
(t/is (= (fs/size path) (:size result)))
(t/is (str/ends-with? (:resource-uri result) (str "/assets/by-id/" rid)))))))
(t/deftest complete-job-multipart-closes-the-render-session
(let [profile (mk-profile)
job-id (mk-job {:status "running"
:profile-id (:id profile)})
session-id (get-in (mgmt :create-job-session {:job-id job-id})
[:result :session-id])
path (artifact-tempfile "penpot")
out (multipart-complete job-id path :session-id session-id)]
(t/is (nil? (:error out)))
(t/is (= {:action :run} (:result out)))
(t/testing "the session is gone, deleted after the commit"
(t/is (nil? (get-session-row session-id))))))
(t/deftest complete-job-multipart-refuses-a-mtype-outside-the-allowlist
(let [profile (mk-profile)
job-id (mk-job {:status "running"
:profile-id (:id profile)})
path (artifact-tempfile "penpot")
out (multipart-complete job-id path
:content {:path path
:filename "result.gif"
:size (fs/size path)
:mtype "image/gif"}
:filename "result.gif"
:mtype "image/gif")]
(t/is (some? (:error out)))
(t/is (= :validation (th/ex-type (:error out))))
(t/is (= :media-type-not-allowed (th/ex-code (:error out))))
(t/testing "no artifact was stored and the row is untouched"
(t/is (zero? (count (th/db-exec! ["SELECT id FROM storage_object"]))))
(t/is (= "running" (:status (get-row job-id)))))))
(t/deftest complete-job-multipart-refuses-a-result-over-the-cap
(let [profile (mk-profile)
job-id (mk-job {:status "running"
:profile-id (:id profile)})
path (artifact-tempfile "penpot")]
(with-mocks [mock {:target 'app.config/get-exporter-max-result-size
:return 0}]
(let [out (multipart-complete job-id path)]
(t/is (some? (:error out)))
(t/is (= :validation (th/ex-type (:error out))))
(t/is (= :request-body-too-large (th/ex-code (:error out))))
(t/testing "no artifact was stored and the row is untouched"
(t/is (zero? (count (th/db-exec! ["SELECT id FROM storage_object"]))))
(t/is (= "running" (:status (get-row job-id)))))))))
(t/deftest complete-job-multipart-of-a-terminal-job-skips-and-stores-nothing
(let [profile (mk-profile)
job-id (mk-job {:status "cancelled"
:profile-id (:id profile)})
path (artifact-tempfile "penpot")
out (multipart-complete job-id path)]
(t/is (nil? (:error out)))
;; the bytes are not used: the completion lost the race with the
;; cancel, and no object is left for the storage GC to find
(t/is (= {:action :skip :status "cancelled"} (:result out)))
(t/is (zero? (count (th/db-exec! ["SELECT id FROM storage_object"]))))))
(t/deftest fail-job-closes-the-render-session
(let [profile (mk-profile)
job-id (mk-job {:status "running"
:profile-id (:id profile)})
session-id (get-in (mgmt :create-job-session {:job-id job-id})
[:result :session-id])
out (mgmt :fail-job {:job-id job-id
:error {:type :internal
:code "processing-error"
:hint "bad image"}
:session-id session-id})]
(t/is (nil? (:error out)))
(t/is (= {:action :run} (:result out)))
(t/is (= "failed" (:status (get-row job-id))))
(t/is (nil? (get-session-row session-id)))))

View File

@ -25,6 +25,18 @@
(def tempfile-types
(conj image-types "application/pdf" "application/zip"))
(def export-artifact-types
"The content types an export job may leave as its result artifact:
what the exporter produces, nothing more. The `complete-job` call of
the management API refuses any other type for the bytes a worker
sends."
#{"application/zip"
"application/pdf"
"image/png"
"image/jpeg"
"image/webp"
"image/svg+xml"})
(def storage-object-types
"Every content-type the system stores in storage objects: uploaded
images and fonts, generated thumbnails, temporary files and the