From db175114ab99a094a7e218c2ab07980517224976 Mon Sep 17 00:00:00 2001 From: Elena Torro Date: Wed, 26 Aug 2026 11:22:41 +0200 Subject: [PATCH] :sparkles: Add export job REST API to exporter --- exporter/src/app/auth.cljs | 98 ++++++++++ exporter/src/app/handlers.cljs | 74 +++++--- exporter/src/app/handlers/export.cljs | 103 ++++++++++ exporter/src/app/handlers/export_frames.cljs | 106 ++++------- exporter/src/app/handlers/export_shapes.cljs | 177 +++++++++--------- exporter/src/app/handlers/jobs.cljs | 60 ++++++ exporter/src/app/http.cljs | 5 +- exporter/src/app/router.cljs | 61 ++++++ .../exporter_tests/export_shapes_test.cljs | 31 +++ exporter/test/exporter_tests/runner.cljs | 12 +- .../test/exporter_tests/scheduler_test.cljs | 81 ++++++++ 11 files changed, 623 insertions(+), 185 deletions(-) create mode 100644 exporter/src/app/auth.cljs create mode 100644 exporter/src/app/handlers/export.cljs create mode 100644 exporter/src/app/handlers/jobs.cljs create mode 100644 exporter/src/app/router.cljs create mode 100644 exporter/test/exporter_tests/export_shapes_test.cljs create mode 100644 exporter/test/exporter_tests/scheduler_test.cljs diff --git a/exporter/src/app/auth.cljs b/exporter/src/app/auth.cljs new file mode 100644 index 0000000000..d6ac0cf042 --- /dev/null +++ b/exporter/src/app/auth.cljs @@ -0,0 +1,98 @@ +;; 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 Sucursal en España SL + +(ns app.auth + "Resolves the caller's session cookie to a real profile id. + + The export commands take `:profile-id` from the request body, which was + harmless while it only picked a pub/sub topic. The job API can read and + cancel other people's work, so its ownership comes from the session: the + token goes to the backend's `get-profile` command, which answers with the + anonymous profile (`uuid/zero`) when it is not a valid session. + + Results are memoized briefly, so a burst of export calls from one client is + one round trip rather than one per request." + (:require + ["undici" :as http] + [app.common.exceptions :as ex] + [app.common.logging :as l] + [app.common.transit :as t] + [app.common.uri :as u] + [app.common.uuid :as uuid] + [app.config :as cf] + [promesa.core :as p])) + +(def ^:private cache-ttl-ms 60000) + +(defonce ^:private cache (atom {})) + +(defn- put-in-cache + "Stores the resolution by session token and removes expired entries. + Without cleanup, long-lived exporters accumulate stale entries" + [cache token profile-id now] + (-> (into {} (remove (fn [[_ {:keys [expires-at]}]] (<= expires-at now))) cache) + (assoc token {:profile-id profile-id + :expires-at (+ now cache-ttl-ms)}))) + +(defn- rpc-uri + [] + (-> (cf/get-internal-uri) + (u/ensure-path-slash) + (u/join "api/rpc/command/get-profile") + (str))) + +(defn- fetch-profile-id + [token] + (let [uri (rpc-uri) + headers #js {"Content-Type" "application/transit+json" + "X-Shared-Key" (str "exporter " cf/management-key) + "Cookie" (str "auth-token=" token)}] + (->> (p/do (http/fetch uri #js {:method "POST" :headers headers :body (t/encode-str {})})) + (p/mcat (fn [^js resp] + (if (= 200 (.-status resp)) + (.text resp) + (p/resolved nil)))) + (p/fmap (fn [body] + (some-> body t/decode-str :id))) + (p/merr (fn [cause] + (l/warn :hint "unable to resolve session profile" :uri uri :cause cause) + (p/resolved nil)))))) + +(defn resolve-profile-id + "Promise of the authenticated profile id, or nil for an anonymous or absent + session." + [token] + (if (nil? token) + (p/resolved nil) + (let [{:keys [profile-id expires-at]} (get @cache token)] + (if (and expires-at (> expires-at (js/Date.now))) + (p/resolved profile-id) + (->> (fetch-profile-id token) + (p/fmap (fn [profile-id] + (let [profile-id (when (and profile-id (not= uuid/zero profile-id)) profile-id)] + (swap! cache put-in-cache token profile-id (js/Date.now)) + profile-id)))))))) + +(defn require-profile-id + "Like `resolve-profile-id`, but rejects anonymous callers." + [token] + (->> (resolve-profile-id token) + (p/mcat (fn [profile-id] + (if profile-id + (p/resolved profile-id) + (ex/raise :type :authentication + :code :authentication-required + :hint "no valid session for this request")))))) + +(defn check-owner! + "Raises unless `profile-id` owns `job`." + [job profile-id] + (when (or (nil? job) + (not= (str (:profile-id job)) (str profile-id))) + (ex/raise :type :not-found + :code :object-not-found + :hint "job does not exist")) + job) diff --git a/exporter/src/app/handlers.cljs b/exporter/src/app/handlers.cljs index b40911fbf1..bd2b453951 100644 --- a/exporter/src/app/handlers.cljs +++ b/exporter/src/app/handlers.cljs @@ -6,22 +6,44 @@ (ns app.handlers (:require + [app.auth :as auth] [app.common.data :as d] - [app.common.exceptions :as ex] [app.common.logging :as l] [app.common.spec :as us] - [app.handlers.export-frames :as export-frames] - [app.handlers.export-shapes :as export-shapes] + [app.handlers.export :as export] [app.util.transit :as t] [clojure.spec.alpha :as s] - [cuerdas.core :as str])) + [promesa.core :as p])) (l/set-level! :debug) +(def ^:private error-codes + #{:queue-full}) + (defn on-error [error exchange] (let [{:keys [type code] :as data} (ex-data error)] (cond + (and (= :validation type) + (contains? error-codes code)) + (let [data {:type :validation + :code code + :hint (ex-message error)}] + (l/warn :hint "rejecting export request" :code code) + (-> exchange + (assoc :response/status 429) + (assoc :response/body (t/encode data)) + (assoc :response/headers {"content-type" "application/transit+json"}))) + + (= :authentication type) + (let [data {:type :authentication + :code code + :hint (ex-message error)}] + (-> exchange + (assoc :response/status 401) + (assoc :response/body (t/encode data)) + (assoc :response/headers {"content-type" "application/transit+json"}))) + (or (= :validation type) (= :assertion type)) (let [explain (us/pretty-explain data) @@ -62,27 +84,27 @@ (assoc :response/body (t/encode (d/without-nils data))) (assoc :response/headers {"content-type" "application/transit+json"})))))) -(defmulti command-spec :cmd) - -(s/def ::id ::us/string) -(s/def ::wait ::us/boolean) -(s/def ::cmd ::us/keyword) - -(defmethod command-spec :export-shapes [_] ::export-shapes/params) -(defmethod command-spec :export-frames [_] ::export-frames/params) - -(s/def ::params - (s/and (s/keys :req-un [::cmd] - :opt-un [::wait]) - (s/multi-spec command-spec :cmd))) - (defn handler - [{:keys [:request/params] :as exchange}] - (let [{:keys [cmd] :as params} (us/conform ::params params)] + "The original `POST /api/export` entry point, and the one the browser backend + still goes through. The export runs as soon as it is asked for, and the + contract is unchanged: `:wait` answers with the finished resource, otherwise + with the resource handle while the work runs." + [{:keys [:request/params :request/auth-token] :as exchange}] + (let [{:keys [cmd wait] :as params} (export/conform-params params)] (l/debug :hint "process-request" :cmd cmd) - (case cmd - :export-shapes (export-shapes/handler exchange params) - :export-frames (export-frames/handler exchange params) - (ex/raise :type :internal - :code :method-not-implemented - :hint (str/istr "method ~{cmd} not implemented"))))) + (->> (auth/resolve-profile-id auth-token) + (p/mcat (fn [profile-id] + ;; The session wins when there is one; the body value stays + ;; the fallback so nothing that used to work stops working. + (export/export! auth-token (cond-> params + (some? profile-id) + (assoc :profile-id profile-id))))) + (p/mcat (fn [{:keys [resource pending]}] + (if wait + (p/fmap (fn [resource] + (assoc exchange :response/body resource)) + pending) + (do + (p/merr (constantly nil) pending) + (p/resolved + (assoc exchange :response/body (dissoc resource :path)))))))))) diff --git a/exporter/src/app/handlers/export.cljs b/exporter/src/app/handlers/export.cljs new file mode 100644 index 0000000000..a7b309f652 --- /dev/null +++ b/exporter/src/app/handlers/export.cljs @@ -0,0 +1,103 @@ +;; 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 Sucursal en España SL + +(ns app.handlers.export + "Handle export jobs" + (:require + [app.common.spec :as us] + [app.handlers.export-frames :as export-frames] + [app.handlers.export-shapes :as export-shapes] + [app.jobs :as jobs] + [app.jobs.scheduler :as scheduler] + [app.jobs.utils :as job.utils] + [clojure.spec.alpha :as s] + [promesa.core :as p])) + +;; --- PARAMS + +(defmulti command-spec :cmd) + +(s/def ::cmd ::us/keyword) +(s/def ::wait ::us/boolean) + +(defmethod command-spec :export-shapes [_] ::export-shapes/params) +(defmethod command-spec :export-frames [_] ::export-frames/params) + +(s/def ::params + (s/and (s/keys :req-un [::cmd] + :opt-un [::wait]) + (s/multi-spec command-spec :cmd))) + +(defn conform-params + [params] + (us/conform ::params params)) + +(defn- prepare + [cmd auth-token params] + (case cmd + :export-shapes (export-shapes/prepare auth-token params) + :export-frames (export-frames/prepare auth-token params))) + +(defn- current + [job] + (or (jobs/lookup (:id job)) job)) + +(defn- run-and-track + [job run] + (->> (p/do (run job)) + (p/mcat (fn [resource] + (->> (jobs/complete! (current job) resource) + (p/fmap (constantly resource))))) + (p/merr (fn [cause] + (if (jobs/cancelled? (:id job)) + (p/rejected cause) + (->> (jobs/fail! (current job) cause) + (p/mcat (fn [_] (p/rejected cause))))))))) + +(defn- run-now! + "Runs the job as soon as it is created, outside the scheduler." + [job] + (->> (p/do (jobs/start! job)) + (p/mcat (fn [job] (p/do ((jobs/run-fn (:id job)) job)))) + (p/fnly (fn [_ _] + (jobs/release! (:id job)) + (job.utils/release! (:id job)))))) + +(defn- create! + [auth-token {:keys [cmd profile-id] :as params} start] + (let [{:keys [resource total headless run]} (prepare cmd auth-token params)] + (->> (jobs/create! {:profile-id profile-id + :cmd cmd + ;; What the renderer will actually do, not what the + ;; client asked for: `is-wasm` alone still renders in + ;; the browser without the `wasm-export` flag, or for + ;; svg, and the backend decides both the admission cap + ;; and whether the client offers to cancel. + :backend (if headless "wasm" "browser") + :total total + :name (:name resource) + :resource-id (:id resource)} + (fn [job] (run-and-track job run))) + (p/fmap (fn [job] + (try + {:job job + :resource resource + :pending (start job)} + (catch :default cause + (jobs/fail! job cause) + (jobs/release! (:id job)) + (throw cause)))))))) + +(defn create-job! + "Returns a promise of `{:job :resource :pending}`." + [auth-token params] + (create! auth-token params scheduler/submit!)) + +(defn export! + "Like `create-job!`, but the work starts right away: This is what + keeps browser exports behaving exactly as they did before there were jobs." + [auth-token params] + (create! auth-token params run-now!)) diff --git a/exporter/src/app/handlers/export_frames.cljs b/exporter/src/app/handlers/export_frames.cljs index f13a389eaa..ed290d4b50 100644 --- a/exporter/src/app/handlers/export_frames.cljs +++ b/exporter/src/app/handlers/export_frames.cljs @@ -6,19 +6,17 @@ (ns app.handlers.export-frames (:require - [app.common.logging :as l] [app.common.spec :as us] - [app.handlers.export-shapes :refer [prepare-exports]] + [app.handlers.export-shapes :refer [count-objects headless-exports? prepare-exports]] [app.handlers.resources :as rsc] - [app.redis :as redis] + [app.jobs :as jobs] + [app.jobs.utils :as job.utils] [app.renderer :as rd] [app.util.shell :as sh] [cljs.spec.alpha :as s] [cuerdas.core :as str] [promesa.core :as p])) -(declare ^:private handle-export) -(declare ^:private create-pdf) (declare ^:private join-pdf) (declare ^:private move-file) @@ -38,85 +36,53 @@ (s/keys :req-un [::exports] :opt-un [::name ::is-wasm])) -(defn handler - [{:keys [:request/auth-token] :as exchange} {:keys [exports] :as params}] - ;; NOTE: we need to have the `:type` prop because the exports - ;; datastructure preparation uses it for creating the groups. - (let [exports (-> (map #(assoc % :type :pdf :scale 1 :suffix "") exports) - (prepare-exports auth-token))] - - (handle-export exchange (assoc params :exports exports)))) - -(defn handle-export - [{:keys [:request/auth-token] :as exchange} {:keys [exports name profile-id is-wasm] :as params}] - (let [topic (str profile-id) - file-id (-> exports first :file-id) - - resource - (rsc/create :pdf (or name (-> exports first :name))) - - on-progress - (fn [done] - (let [data {:type :export-update - :resource-id (:id resource) - :status "running" - :done done}] - (redis/pub! topic data))) - - on-complete - (fn [resource] - (let [data {:type :export-update - :resource-id (:id resource) - :resource-uri (:uri resource) - :name (:name resource) - :filename (:filename resource) - :mtype (:mtype resource) - :status "ended"}] - (redis/pub! topic data))) - - on-error - (fn [cause] - (l/error :hint "unexpected error on frames exportation" :cause cause) - (let [data {:type :export-update - :resource-id (:id resource) - :name (:name resource) - :filename (:filename resource) - :status "error" - :cause (ex-message cause)}] - (redis/pub! topic data))) - - result-cache - (atom []) +(defn- run-export + [job auth-token resource {:keys [exports is-wasm file-id]}] + (let [rendered (atom []) on-object - (fn [{:keys [path] :as object}] - (let [res (swap! result-cache conj path)] - (on-progress (count res)))) + (fn [{:keys [path] :as _object}] + (job.utils/track! (:id job) path) + (jobs/progress! job (count (swap! rendered conj path)))) - procs - (->> (seq exports) - (map #(rd/render (assoc % :is-wasm is-wasm) on-object)))] + exports + (map #(assoc % :is-wasm is-wasm :job-id (:id job)) exports)] - (->> (p/all procs) - (p/fmap (fn [] @result-cache)) - (p/mcat (partial join-pdf file-id)) + (job.utils/track! (:id job) (:path resource)) + (->> (rd/with-scope exports + (fn [render] + (jobs/check-cancelled! job) + (->> exports + (map (fn [export] (render export on-object))) + (p/all)))) + (p/fmap (fn [_] @rendered)) + (p/mcat (partial join-pdf job file-id)) (p/mcat (partial move-file resource)) (p/fmap (constantly resource)) (p/mcat (partial rsc/upload-resource auth-token)) (p/mcat (fn [resource] (->> (sh/stat (:path resource)) (p/fmap #(merge resource %))))) - (p/merr on-error) - (p/fnly (fn [resource cause] - (when-not cause - (on-complete resource))))) + (p/fmap (fn [resource] (dissoc resource :path)))))) - (assoc exchange :response/body (dissoc resource :path)))) +(defn prepare + [auth-token {:keys [exports name is-wasm] :as _params}] + (let [exports (-> (map #(assoc % :type :pdf :scale 1 :suffix "") exports) + (prepare-exports auth-token is-wasm)) + resource (rsc/create :pdf (or name (-> exports first :name))) + file-id (-> exports first :file-id)] + {:resource resource + :total (count-objects exports) + :headless (headless-exports? exports is-wasm) + :run (fn [job] (run-export job auth-token resource + {:exports exports + :is-wasm is-wasm + :file-id file-id}))})) (defn- join-pdf - [file-id paths] + [job file-id paths] (p/let [prefix (str/concat "penpot.pdfunite." file-id ".") - path (sh/tempfile :prefix prefix :suffix ".pdf")] + path (job.utils/track! (:id job) (sh/tempfile :prefix prefix :suffix ".pdf"))] (apply sh/run-cmd! "pdfunite" (conj (vec paths) path)) path)) diff --git a/exporter/src/app/handlers/export_shapes.cljs b/exporter/src/app/handlers/export_shapes.cljs index a07b58ad72..954285ba1e 100644 --- a/exporter/src/app/handlers/export_shapes.cljs +++ b/exporter/src/app/handlers/export_shapes.cljs @@ -7,10 +7,10 @@ (ns app.handlers.export-shapes (:require [app.common.data :as d] - [app.common.logging :as l] [app.common.spec :as us] [app.handlers.resources :as rsc] - [app.redis :as redis] + [app.jobs :as jobs] + [app.jobs.utils :as job.utils] [app.renderer :as rd] [app.util.mime :as mime] [app.util.shell :as sh] @@ -18,9 +18,6 @@ [cuerdas.core :as str] [promesa.core :as p])) -(declare ^:private handle-single-export) -(declare ^:private handle-multiple-export) -(declare ^:private assoc-file-name) (declare prepare-exports) ;; Regex to clean namefiles @@ -50,87 +47,92 @@ (s/keys :req-un [::exports ::profile-id] :opt-un [::wait ::name ::skip-children ::force-multiple ::is-wasm])) -(defn handler - [{:keys [:request/auth-token] :as exchange} {:keys [exports force-multiple] :as params}] - (let [exports (prepare-exports exports auth-token)] - (if (and (not force-multiple) - (= 1 (count exports)) - (= 1 (count (-> exports first :objects)))) - (handle-single-export exchange (-> params - (assoc :export (first exports)) - (dissoc :exports))) - (handle-multiple-export exchange (assoc params :exports exports))))) +(defn count-objects + [exports] + (reduce + 0 (map (comp count :objects) exports))) -(defn- handle-single-export - [{:keys [:request/auth-token] :as exchange} {:keys [export name skip-children is-wasm] :as params}] - (let [resource (rsc/create (:type export) (or name (:name export))) - export (assoc export :skip-children skip-children :is-wasm (boolean is-wasm))] +(defn- render! + [job export on-object] + (jobs/check-cancelled! job) + (rd/render (assoc export :job-id (:id job)) on-object)) - (->> (rd/render export - (fn [{:keys [path] :as object}] - (sh/move! path (:path resource)))) +(defn- scoped-renders + "Renders every export, the headless ones sharing a single worker." + [job exports on-object] + (rd/with-scope exports + (fn [render] + (jobs/check-cancelled! job) + (->> exports + (map (fn [export] (render export on-object))) + (p/all))))) + +(defn- run-single + [job auth-token resource {:keys [export is-wasm skip-children]}] + (job.utils/track! (:id job) (:path resource)) + (->> (render! job + (assoc export :skip-children skip-children :is-wasm (boolean is-wasm)) + (fn [{:keys [path] :as _object}] + (job.utils/track! (:id job) path) + (sh/move! path (:path resource)))) + (p/fmap (constantly resource)) + (p/mcat (partial rsc/upload-resource auth-token)) + (p/fmap (fn [resource] (dissoc resource :path))))) + +(defn- run-multiple + [job auth-token resource {:keys [exports is-wasm]}] + (let [failure (volatile! nil) + + zip (rsc/create-zip :resource resource + :on-error (fn [cause] (vreset! failure cause)) + :on-progress (fn [{:keys [done]}] + (jobs/progress! job done))) + + append (fn [{:keys [filename path] :as _object}] + (job.utils/track! (:id job) path) + (rsc/add-to-zip zip path (str/replace filename sanitize-file-regex "_")))] + + (job.utils/track! (:id job) (:path resource)) + (->> (scoped-renders job + (map #(assoc % :is-wasm (boolean is-wasm) :job-id (:id job)) exports) + append) + (p/mcat (fn [_] + (if-let [cause @failure] + (p/rejected cause) + (rsc/close-zip zip)))) (p/fmap (constantly resource)) (p/mcat (partial rsc/upload-resource auth-token)) - (p/fmap (fn [resource] - (dissoc resource :path))) - (p/fmap (fn [resource] - (assoc exchange :response/body resource))) - (p/merr (fn [cause] - (l/error :hint "unexpected error on single export" - :cause cause) - (p/rejected cause)))))) + (p/fmap (fn [resource] (dissoc resource :path)))))) -(defn- handle-multiple-export - [{:keys [:request/auth-token] :as exchange} {:keys [exports wait profile-id name is-wasm] :as params}] - (let [resource (rsc/create :zip (or name (-> exports first :name))) - total (count exports) - topic (str profile-id) +(defn headless-exports? + "Whether any of `exports` renders headless, and so whether the job leases a + render worker. Mirrors what `rd/with-scope` decides at run time." + [exports is-wasm] + (boolean (some #(rd/headless? {:is-wasm is-wasm :type (:type %)}) exports))) - on-progress (fn [{:keys [done]}] - (when-not wait - (let [data {:type :export-update - :resource-id (:id resource) - :status "running" - :total total - :done done}] - (redis/pub! topic data)))) +(defn prepare + [auth-token {:keys [exports force-multiple name skip-children is-wasm] :as _params}] + (let [exports (prepare-exports exports auth-token is-wasm) + headless? (headless-exports? exports is-wasm) + single? (and (not force-multiple) + (= 1 (count exports)) + (= 1 (count (-> exports first :objects))))] + (if single? + (let [export (first exports) + resource (rsc/create (:type export) (or name (:name export)))] + {:resource resource + :total 1 + :headless headless? + :run (fn [job] (run-single job auth-token resource + {:export export + :is-wasm is-wasm + :skip-children skip-children}))}) - on-error (fn [cause] - (l/error :hint "unexpected error on multiple export" :cause cause) - (if wait - (p/rejected cause) - (redis/pub! topic {:type :export-update - :resource-id (:id resource) - :status "error" - :cause (ex-message cause)}))) - - zip (rsc/create-zip :resource resource - :on-error on-error - :on-progress on-progress) - - append (fn [{:keys [filename path] :as resource}] - (rsc/add-to-zip zip path (str/replace filename sanitize-file-regex "_"))) - - proc (->> exports - (map (fn [export] (rd/render (assoc export :is-wasm (boolean is-wasm)) append))) - (p/all) - (p/mcat (fn [_] (rsc/close-zip zip))) - (p/fmap (constantly resource)) - (p/mcat (partial rsc/upload-resource auth-token)) - (p/fmap (fn [resource] - (let [data {:type :export-update - :name (:name resource) - :filename (:filename resource) - :resource-id (:id resource) - :resource-uri (:uri resource) - :mtype (:mtype resource) - :status "ended"}] - (p/do (redis/pub! topic data) - (assoc exchange :response/body resource))))) - (p/merr on-error))] - (if wait - (p/then proc #(assoc exchange :response/body (dissoc % :path))) - (assoc exchange :response/body (dissoc resource :path))))) + (let [resource (rsc/create :zip (or name (-> exports first :name)))] + {:resource resource + :total (count-objects exports) + :headless headless? + :run (fn [job] (run-multiple job auth-token resource + {:exports exports :is-wasm is-wasm}))})))) (defn- assoc-file-name "A transducer that assocs a candidate filename and avoid duplicates" @@ -160,13 +162,18 @@ default-partition-size 50) (defn prepare-exports - [exports token] - (letfn [(process-group [group] - (sequence (comp (partition-all default-partition-size) - (map process-partition)) - group)) + [exports token is-wasm] + (letfn [(process-group [[part1 :as group]] + ;; The browser renders a partition as a single DOM page, so it is + ;; chunked to bound that page. A wasm export is headless, so + ;; it does not need to be chunked, and can be rendered as a single partition. + (if (rd/headless? {:is-wasm is-wasm :type (:type part1)}) + [(build-render group)] + (sequence (comp (partition-all default-partition-size) + (map build-render)) + group))) - (process-partition [[part1 :as part]] + (build-render [[part1 :as part]] {:file-id (:file-id part1) :page-id (:page-id part1) :share-id (:share-id part1) diff --git a/exporter/src/app/handlers/jobs.cljs b/exporter/src/app/handlers/jobs.cljs new file mode 100644 index 0000000000..eca18e8cca --- /dev/null +++ b/exporter/src/app/handlers/jobs.cljs @@ -0,0 +1,60 @@ +;; 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 Sucursal en España SL + +(ns app.handlers.jobs + "REST surface for export jobs, under `/api/export/jobs`. + + Ownership always comes from the session (see `app.auth`), never from the + request body." + (:require + [app.auth :as auth] + [app.common.exceptions :as ex] + [app.common.logging :as l] + [app.handlers.export :as export] + [app.jobs :as jobs] + [promesa.core :as p])) + +(defn create + [{:keys [:request/auth-token :request/params] :as exchange}] + (->> (auth/require-profile-id auth-token) + (p/mcat (fn [profile-id] + (let [params (-> params + (assoc :profile-id profile-id) + (export/conform-params))] + (l/dbg :hint "create export job" :cmd (:cmd params) :profile-id (str profile-id)) + (export/create-job! auth-token params)))) + (p/fmap (fn [{:keys [job resource pending]}] + ;; A failure is reported through the job record, so the + ;; promise must not surface as an unhandled rejection. + (p/merr (constantly nil) pending) + (-> exchange + (assoc :response/body (-> (or (jobs/lookup (:id job)) job) + (assoc :filename (:filename resource)) + (assoc :mtype (:mtype resource))))))))) + +(defn fetch + [{:keys [:request/auth-token] :as exchange} job-id] + (->> (auth/require-profile-id auth-token) + (p/mcat (fn [profile-id] + (->> (jobs/fetch job-id) + (p/fmap #(auth/check-owner! % profile-id))))) + (p/fmap (fn [job] + (assoc exchange :response/body job))))) + +(defn cancel + [{:keys [:request/auth-token] :as exchange} job-id] + (->> (auth/require-profile-id auth-token) + (p/mcat (fn [profile-id] + (->> (jobs/fetch job-id) + (p/fmap #(auth/check-owner! % profile-id))))) + (p/mcat (fn [job] (jobs/cancel! (:id job)))) + (p/mcat (fn [_] (jobs/fetch job-id))) + (p/fmap (fn [job] + (if job + (assoc exchange :response/body job) + (ex/raise :type :not-found + :code :object-not-found + :hint "job does not exist")))))) diff --git a/exporter/src/app/http.cljs b/exporter/src/app/http.cljs index e025ca2584..93100885f0 100644 --- a/exporter/src/app/http.cljs +++ b/exporter/src/app/http.cljs @@ -15,6 +15,7 @@ [app.common.transit :as t] [app.config :as cf] [app.handlers :as handlers] + [app.router :as router] [cuerdas.core :as str] [lambdaisland.uri :as u] [promesa.core :as p])) @@ -94,7 +95,7 @@ size (js/Buffer.byteLength data "utf-8")] (-> exchange (assoc :response/body data) - (assoc :response/status 200) + (assoc :response/status (or status 200)) (update :response/headers assoc "content-type" "application/transit+json") (update :response/headers assoc "content-length" size))) @@ -159,7 +160,7 @@ (defn init [] - (let [handler (-> handlers/handler + (let [handler (-> (router/create handlers/handler) (wrap-health) (wrap-auth "auth-token") (wrap-response-format) diff --git a/exporter/src/app/router.cljs b/exporter/src/app/router.cljs new file mode 100644 index 0000000000..991ab60a24 --- /dev/null +++ b/exporter/src/app/router.cljs @@ -0,0 +1,61 @@ +;; 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 Sucursal en España SL + +(ns app.router + "Method + path dispatch. + + Requests arrive with whatever prefix the proxy in front uses (`/api/export` + in devenv, `/` when talking to the process directly), so routes are matched + on the remainder after that prefix." + (:require + [app.common.exceptions :as ex] + [app.handlers.jobs :as jobs.handlers] + [cuerdas.core :as str])) + +(def ^:private mount-point "/api/export") + +(defn- route-path + [path] + (let [path (or path "/") + path (if (str/starts-with? path mount-point) + (subs path (count mount-point)) + path) + path (str/rtrim path "/")] + (if (str/empty? path) "/" path))) + +(defn- job-id + [path prefix] + (let [id (subs path (count prefix))] + (when-not (or (str/empty? id) (str/includes? id "/")) + id))) + +(defn create + "Builds the request handler. `legacy-handler` serves the original + `POST /api/export` command multiplex." + [legacy-handler] + (fn [{:keys [:request/method :request/path] :as exchange}] + (let [path (route-path path)] + (cond + (and (= "post" method) (= "/" path)) + (legacy-handler exchange) + + (and (= "post" method) (= "/jobs" path)) + (jobs.handlers/create exchange) + + (and (= "get" method) (str/starts-with? path "/jobs/")) + (if-let [id (job-id path "/jobs/")] + (jobs.handlers/fetch exchange id) + (ex/raise :type :not-found :code :object-not-found :hint "unknown route")) + + (and (= "delete" method) (str/starts-with? path "/jobs/")) + (if-let [id (job-id path "/jobs/")] + (jobs.handlers/cancel exchange id) + (ex/raise :type :not-found :code :object-not-found :hint "unknown route")) + + :else + (ex/raise :type :not-found + :code :route-not-found + :hint (str "no route for " method " " path)))))) diff --git a/exporter/test/exporter_tests/export_shapes_test.cljs b/exporter/test/exporter_tests/export_shapes_test.cljs new file mode 100644 index 0000000000..d512cabd78 --- /dev/null +++ b/exporter/test/exporter_tests/export_shapes_test.cljs @@ -0,0 +1,31 @@ +;; 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 Sucursal en España SL + +(ns exporter-tests.export-shapes-test + "Chunking of the browser backend." + (:require + [app.common.uuid :as uuid] + [app.handlers.export-shapes :as export-shapes] + [cljs.test :as t :include-macros true])) + +(defn- exports + [n type scale] + (let [file-id (uuid/next) + page-id (uuid/next)] + (mapv (fn [i] + {:file-id file-id + :page-id page-id + :object-id (uuid/next) + :name (str "shape-" i) + :suffix "" + :scale scale + :type type}) + (range n)))) + +(t/deftest browser-exports-are-chunked + (let [parts (export-shapes/prepare-exports (exports 120 :png 1) "token" false)] + (t/is (= 3 (count parts))) + (t/is (= [50 50 20] (mapv (comp count :objects) parts))))) diff --git a/exporter/test/exporter_tests/runner.cljs b/exporter/test/exporter_tests/runner.cljs index a0b7c3020d..82c3e4c46f 100644 --- a/exporter/test/exporter_tests/runner.cljs +++ b/exporter/test/exporter_tests/runner.cljs @@ -10,15 +10,23 @@ [cljs.test :as t] [clojure.string :as str] [clojure.tools.cli :refer [parse-opts]] + [exporter-tests.export-shapes-test] + [exporter-tests.jobs-test] [exporter-tests.renderer-svg-test] + [exporter-tests.scheduler-test] [exporter-tests.shell-test] + [exporter-tests.wasm-pool-test] [goog.object :as gobj])) (enable-console-print!) (def test-namespaces - ['exporter-tests.renderer-svg-test - 'exporter-tests.shell-test]) + ['exporter-tests.export-shapes-test + 'exporter-tests.jobs-test + 'exporter-tests.renderer-svg-test + 'exporter-tests.scheduler-test + 'exporter-tests.shell-test + 'exporter-tests.wasm-pool-test]) (assert (every? find-ns-obj test-namespaces) "test-namespaces contains a namespace that isn't required in runner.cljs") diff --git a/exporter/test/exporter_tests/scheduler_test.cljs b/exporter/test/exporter_tests/scheduler_test.cljs new file mode 100644 index 0000000000..48c3791632 --- /dev/null +++ b/exporter/test/exporter_tests/scheduler_test.cljs @@ -0,0 +1,81 @@ +;; 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 Sucursal en España SL + +(ns exporter-tests.scheduler-test + "Admission control. A headless job leases one render worker for its whole run, + so no more of them may start than there are workers." + (:require + [app.common.uuid :as uuid] + [app.jobs :as jobs] + [app.jobs.scheduler :as scheduler] + [app.wasm.pool :as pool] + [cljs.test :as t :include-macros true] + [promesa.core :as p])) + +(defn- pin-capacity! + "Fixes the worker count for the test. Returns the thunk that puts it back; + `with-redefs` cannot be used here, since the scheduler keeps admitting jobs + after the body of the test has returned." + [n] + (let [original pool/capacity] + (set! pool/capacity (constantly n)) + (fn [] (set! pool/capacity original)))) + +(defn- create! + [backend run-fn] + (jobs/create! {:profile-id (uuid/next) + :cmd :export-shapes + :backend backend + :total 1 + :name "test" + :resource-id (uuid/next)} + run-fn)) + +(defn- gate + "A promise and the fn that settles it, standing in for a render in flight." + [] + (let [resolve* (volatile! nil) + pending (p/create (fn [resolve _] (vreset! resolve* resolve)))] + [pending (fn [] (@resolve* nil))])) + +(t/deftest headless-jobs-wait-for-a-render-worker + (t/testing "with one worker, the second headless job stays queued until the first ends" + (t/async done + (let [restore! (pin-capacity! 1) + [render open] (gate) + started (atom [])] + (p/let [job1 (create! "wasm" (fn [_] (swap! started conj :one) render)) + job2 (create! "wasm" (fn [_] (swap! started conj :two) (p/resolved nil)))] + (let [p1 (scheduler/submit! job1) + p2 (scheduler/submit! job2)] + (p/do + (p/delay 10) + (t/is (= [:one] @started)) + (t/is (= "running" (:state (jobs/lookup (:id job1))))) + (t/is (= "queued" (:state (jobs/lookup (:id job2))))) + (open) + (p/all [p1 p2]) + (t/is (= [:one :two] @started)) + (restore!) + (done)))))))) + +(t/deftest a-browser-job-is-not-held-back-by-the-worker-pool + (t/testing "the headless cap applies to headless jobs only" + (t/async done + (let [restore! (pin-capacity! 1) + [render open] (gate) + started (atom [])] + (p/let [job1 (create! "wasm" (fn [_] (swap! started conj :wasm) render)) + job2 (create! "browser" (fn [_] (swap! started conj :browser) (p/resolved nil)))] + (let [p1 (scheduler/submit! job1) + p2 (scheduler/submit! job2)] + (p/do + (p/delay 10) + (t/is (= [:wasm :browser] @started)) + (open) + (p/all [p1 p2]) + (restore!) + (done))))))))