✨ Add export job REST API to exporter

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)

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))))))))))

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!))

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))
 

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)

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"))))))

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)

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))))))

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)))))

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")

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))))))))