From 8328974f2e668042752b4acb453da95e819c6a4f Mon Sep 17 00:00:00 2001 From: Andrey Antukh Date: Wed, 7 Oct 2026 07:42:48 +0000 Subject: [PATCH] :sparkles: 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 --- backend/src/app/config.clj | 12 + backend/src/app/rpc/management/jobs.clj | 214 +++++++++++++++--- .../rpc_management_jobs_test.clj | 157 ++++++++++++- common/src/app/common/media.cljc | 12 + 4 files changed, 356 insertions(+), 39 deletions(-) diff --git a/backend/src/app/config.clj b/backend/src/app/config.clj index dfc4bfd23e..74f011a02c 100644 --- a/backend/src/app/config.clj +++ b/backend/src/app/config.clj @@ -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] diff --git a/backend/src/app/rpc/management/jobs.clj b/backend/src/app/rpc/management/jobs.clj index 6f069fcd55..2bf2d21293 100644 --- a/backend/src/app/rpc/management/jobs.clj +++ b/backend/src/app/rpc/management/jobs.clj @@ -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 diff --git a/backend/test/backend_tests/rpc_management_jobs_test.clj b/backend/test/backend_tests/rpc_management_jobs_test.clj index 40b91a5314..0ec31e30dc 100644 --- a/backend/test/backend_tests/rpc_management_jobs_test.clj +++ b/backend/test/backend_tests/rpc_management_jobs_test.clj @@ -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))))) diff --git a/common/src/app/common/media.cljc b/common/src/app/common/media.cljc index ea0c11b530..880c07b7a6 100644 --- a/common/src/app/common/media.cljc +++ b/common/src/app/common/media.cljc @@ -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