github/penpot
1
0
Fork 0
mirror of https://github.com/penpot/penpot.git synced 2026-10-30 06:20:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

✨ Add the exporter worker pollers and the export jobs run

Browse Source 
The `:exporter` queue gets its consumer: K pollers (the new
`:exporter-worker-concurrency`), each one with its own Redis
connection, so the blocking BLPOP is both the semaphore of the slot and
the backpressure: a poller claims one job and does not pop again until
the job settles. The dispatcher's `[job-id, scheduled-at]` payloads are
read per pop; a corrupt payload is logged and dropped, a Redis hiccup
is a failed turn that the loop retries after a small delay, and no
test exercises the forever loop directly.

The run of the `:export-assets` job ports the legacy export machinery
to the substrate: the item transducers and partitioning of
export-shapes, the zip assembly and the pdfunite join of export-frames,
rendering as the session the backend minted. The progress contract
dresses the beats (`:preparing`, `:rendering`, `:packaging` with the
`objects` and `pages` counters, throttled per object) and the settle is
one `complete-job` multipart that stores the artifact and closes the
session; an export failure settles as `fail-job` with the session id,
self-contained, so the loop only ever sees a resolved run.

AI-assisted-by: glm-5.3-flash
This commit is contained in:
Andrey Antukh 2026-10-07 09:08:53 +00:00
parent fe636282cb
commit c0b4430293
4 changed files with 653 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

262
exporter/src/app/consumer/exports.cljs Normal file
View File

@ -0,0 +1,262 @@
;; 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 SUBSIDIARY SL
(ns app.consumer.exports
"The runner of the `:export-assets` job.
Takes the params the job-def froze and transplant the legacy export
machinery into the jobs substrate: the name transducers and
partitioning of `app.handlers.export-shapes`, and the pdf assembly
its `export-frames` twin does, dress the same run the old `/api/export`
served. What changes around them is the frame: the session minted
through `create-job-session` renders as the user, the milestone
vocabulary of the progress contract dresses the beats (`:preparing`,
`:rendering`, `:packaging`, with the `objects` and `pages` counters
of every partition), and the settle is one `complete-job` multipart
call that stores the artifact and closes the session in the same
step.
The run never rejects on an export failure: whatever goes wrong
reaches the backend as `fail-job` (with the session id, so the render
session does not outlive the settle). Only bugs of this process
reject."
(:require
[app.common.exceptions :as ex]
[app.common.logging :as l]
[app.consumer.api :as api]
[app.handlers.export-shapes :as shapes]
[app.handlers.resources :as rsc]
[app.jobs.utils :as job.utils]
[app.renderer :as rd]
[app.util.shell :as sh]
[cuerdas.core :as str]
[promesa.core :as p]))
(def ^:private report-throttle-ms 250)
;; ---- PROGRESS (the milestones of the contract)
(defn- report!
[job-id stage counters]
(api/report-job-progress job-id {:stage stage :counters counters}))
(defn- start-beats!
"One progress reporter per run. The stage changes beat by force; the
per-object beats are throttled the way the legacy export was, since
hundreds of objects would otherwise be hundreds of malformed calls
for information nobody reads at that resolution."
[job-id]
(let [last-beat (atom 0)]
(fn [stage counters & {:keys [force?]}]
(if force?
(report! job-id stage counters)
(let [now (js/Date.now)]
(when (>= (- now @last-beat) report-throttle-ms)
(reset! last-beat now)
(report! job-id stage counters)))))))
(defn- counter
"One counter of the vocabulary: `kind` is `objects` or `pages`."
[kind current total]
{kind {:current current :total total}})
;; ---- THE PARAMS
(defn- frame-item?
"The frames export names pages of a file without a `type`: the pdf
of their pages. A shape item always names its type."
[item]
(nil? (:type item)))
(defn- normalize-items
"The items the claim delivered are the plain JSON the job row holds:
the type arrives as text and the renderer dispatches on the keywords
the legacy surface received. The frames items are pages, not typed
shapes: they name the pdf of a page each."
[items]
(->> items
(mapv (fn [item]
(cond-> item
(string? (:type item)) (update :type keyword)
(nil? (:type item)) (assoc :type :pdf :scale 1 :suffix ""))))))
(defn- make-plan
"The render plan the legacy handlers prepare: the same transducers of
names and partition size they served the old surface with, the same
`single?` rule (one prepared export with one object, not forced),
and the artifact the whole run fills."
[token {:keys [exports force-multiple name skip-children is-wasm]}]
(let [items (normalize-items exports)
frames? (boolean (every? frame-item? items))
prepared (shapes/prepare-exports items token is-wasm)
single? (and (not frames?)
(not (true? force-multiple))
(= 1 (count prepared))
(= 1 (count (-> prepared first :objects))))
kind (cond
single? (-> prepared first :type)
frames? :pdf
:else :zip)]
{:frames? frames?
:single? single?
:prepared prepared
:total (shapes/count-objects prepared)
:skip-children skip-children
:is-wasm (boolean is-wasm)
:counter-kind (if frames? :pages :objects)
:resource (rsc/create kind
(or name (-> prepared first :name)))}))
;; ---- THE RUNNERS
(defn- run-single!
"One render, one object: the artifact IS the object, moved into the
resource path the multipart settle will name."
[progress! plan]
(let [{:keys [job-id resource counter-kind]} plan
export (-> plan :prepared first)
object (atom nil)]
(job.utils/track! job-id (:path resource))
(p/let [_ (progress! :rendering (counter counter-kind 0 1) :force? true)
_ (rd/render (assoc export
:job-id job-id
:skip-children (:skip-children plan))
(fn [obj]
(reset! object obj)
(job.utils/track! job-id (:path obj))))
_ (sh/move! (:path @object) (:path resource))
_ (progress! :packaging (counter counter-kind 1 1) :force? true)]
resource)))
(defn- run-multiple!
"The multi-object render: every prepared export renders with the same
on-object callback, appending each object to the zip as it lands, and
the progress counts what has landed so far. Any zipping error
surfaces after the renders, so the failures of the writer do not
compete with the ones of the render."
[progress! plan]
(let [{:keys [job-id resource prepared total counter-kind]} plan
failure (volatile! nil)
rendered (volatile! 0)
zip (rsc/create-zip :resource resource
:on-error (fn [cause] (vreset! failure cause))
:on-progress (fn [_] nil))
append (fn [{:keys [filename path]}]
(job.utils/track! job-id path)
(vswap! rendered inc)
(progress! :rendering
(counter counter-kind @rendered total))
(rsc/add-to-zip zip path
(str/replace filename
shapes/sanitize-file-regex
"_")))]
(p/let [_ (progress! :rendering (counter counter-kind 0 total) :force? true)
_ (rd/with-scope prepared
(fn [scoped-render]
(p/all (map (fn [export]
(scoped-render
(assoc export :job-id job-id)
append))
prepared))))
error (if-let [cause @failure]
(p/rejected cause)
(rsc/close-zip zip))
_ (do error)
_ (progress! :packaging (counter counter-kind total total) :force? true)]
resource)))
(defn- join-pdf!
[job-id file-id paths]
(let [path (job.utils/track! job-id
(sh/tempfile :prefix (str/concat "penpot.pdfunite." file-id ".")
:suffix ".pdf"))]
(p/let [_ (sh/run-cmd! "pdfunite" (into [] (concat (vec paths) [path])))]
path)))
(defn- run-frames!
"The frames render: a file per page, joined into the pdf of the file
once every page has landed."
[progress! plan]
(let [{:keys [job-id resource prepared total]} plan
file-id (-> prepared first :file-id)
paths (volatile! [])
rendered (volatile! 0)
on-object (fn [{:keys [path]}]
(job.utils/track! job-id path)
(vswap! paths conj path)
(vswap! rendered inc)
(progress! :rendering
(counter :pages @rendered total)))]
(p/let [_ (progress! :rendering (counter :pages 0 total) :force? true)
_ (rd/with-scope prepared
(fn [scoped-render]
(p/all (map (fn [export]
(scoped-render
(assoc export :job-id job-id :is-wasm (:is-wasm plan))
on-object))
prepared))))
joined (join-pdf! job-id file-id @paths)
_ (sh/move! joined (:path resource))
_ (progress! :packaging (counter :pages total total) :force? true)]
resource)))
;; ---- THE SETTLE
(defn- failure-report
"The error shape the backend stores for `job.error`: type and code
keywords, hint the human message. A cause that is not a penpot error
becomes an internal one, with only its message as the hint."
[cause]
(let [data (ex-data cause)]
{:type (or (:type data) :internal)
:code (or (:code data) :export-failed)
:hint (or (:hint data) (ex-message cause) "export failed")}))
(defn- settle-failure!
"Settles the job as failed, after releasing the temp files the run
owns. The settle never rejects: if the failure cannot reach the
backend, the row keeps running until the lease GC turns it `aborted`,
which is the honest state for a worker that cannot talk."
[job-id session cause]
(p/do
(job.utils/release! job-id)
(->> (api/fail-job job-id (failure-report cause)
:session-id (some-> @session :session-id))
(p/merr (fn [settle-cause]
(l/warn :hint "unable to settle the failed job"
:job-id job-id :cause settle-cause))))))
(defn- run-prepared!
[beats! plan]
(cond
(:single? plan) (run-single! beats! plan)
(:frames? plan) (run-frames! beats! plan)
:else (run-multiple! beats! plan)))
(defn run-export!
[{:keys [job-id] :as _job} params]
(let [session (atom nil)
beats! (start-beats! job-id)]
(p/catch
(p/let [session' (api/create-job-session job-id)
_ (reset! session session')
token (:session-token session')
_ (beats! :preparing {})
plan (make-plan token params)
resource (run-prepared! beats! plan)
artifact {:path (str (:path resource))
:filename (:filename resource)
:mtype (:mtype resource)}
_ (api/complete-job-with-artifact
{:job-id job-id :session-id (:session-id session')}
artifact)
_ (job.utils/release! job-id)
_ (l/info :hint "export job settled" :job-id job-id
:outcome "completed")]
nil)
(fn [cause]
(l/error :hint "export job failed" :job-id job-id :cause cause)
(settle-failure! job-id session cause)))))

128
exporter/src/app/consumer/worker.cljs Normal file
View File

@ -0,0 +1,128 @@
;; 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 SUBSIDIARY SL
(ns app.consumer.worker
"The consumer of the backend `:exporter` queue.
K pollers run in parallel, each one owning its Redis connection: the
blocking BLPOP of a poller holds that connection for the whole wait,
and the poller does not come back to the queue until the job it
claimed has settled, so the connection is both the semaphore of the
slot and the backpressure — nothing the queue receives can appear
twice taken, and no process ever runs more exports than it has
connections.
The dispatcher pushes `[job-id, scheduled-at]` as JSON into
`penpot.worker.queue:<tenant>:exporter`. Corrupt payloads are logged
and dropped; a Redis hiccup is a lost turn, not a lost job: the
poller waits a bit and pops again (what was already delivered is
claimed, and the loop does not stop the process)."
(:require
["ioredis" :as redis]
[app.common.logging :as l]
[app.common.uuid :as uuid]
[app.config :as cf]
[app.consumer.api :as api]
[app.consumer.config :as ccfg]
[app.consumer.exports :as exports]
[promesa.core :as p]))
(def ^:private poll-timeout-s 5)
(def ^:private reconnect-delay-ms 1000)
(defonce ^:private pollers (atom {}))
(defn- queue-key
"The list the backend dispatcher pushes to."
[]
(ccfg/queue-key))
(defn- connect!
"One Redis connection of one poller. ioredis reconnects on its own
strategy, so a fallen Redis comes back with this process doing
nothing but wait (and the failed pop comes right after)."
[]
(new redis/default (cf/get :redis-uri)))
;; ---- THE QUEUE
(defn read-payload!
"The JSON payload of the dispatcher: `job-id` and the `scheduled-at`
of the row when it was pushed. Anything else is a corrupt payload:
none, and the caller drops it."
[payload]
(try
(let [[job-id scheduled-at] (js->clj (js/JSON.parse payload))]
{:job-id job-id :scheduled-at scheduled-at})
(catch :default cause
(l/warn :hint "corrupt queue payload dropped"
:payload (str payload) :cause cause)
nil)))
(defn process!
"One payload of the queue: claim the job, and when the backend gives
it to this poller, run the export to its settle. `skip` means
something else won it (or it is gone): the payload is dropped, with
the state the backend says."
[{:keys [job-id scheduled-at]}]
(p/let [answer (api/claim-job job-id scheduled-at)]
(if (= :skip (:action answer))
(l/info :hint "skipping job" :job-id (str job-id)
:name (:name answer) :status (:status answer))
(p/do
(l/info :hint "running job" :job-id (str job-id) :name (:name answer))
(exports/run-export! {:job-id job-id} (:params answer))
(l/info :hint "job settled" :job-id (str job-id))))))
;; ---- THE POLLER
(defn poll-once!
"One turn: pop the queue, claim what came, and wait for the settle.
An empty pop and a dropped payload both resolve; a claimed job is
awaited the whole run."
[conn key on-payload]
(p/let [answer (.blpop ^js conn key poll-timeout-s)
payload (second answer)
claim (when (some? payload) (read-payload! payload))
settled (when (some? claim) (on-payload claim))]
settled))
(defn- poll-loop!
"The wait of one poller, turn after turn, for the life of the
process. A failed turn (Redis down, a bug) is logged, never allowed
to end the loop; a delay keeps the retry from spinning while the
connection rebuilds."
[conn key on-payload]
(p/loop []
(->> (poll-once! conn key on-payload)
(p/merr (fn [cause]
(l/error :hint "poller turn failed" :key key :cause cause)
(p/delay reconnect-delay-ms)))
(p/fmap (fn [_] (p/recur))))))
;; ---- THE START
(defn start!
"Launches `K` pollers, one connection each. The worker cleans the
temp files a previous process left behind at the same moment: a
crashed worker owns nothing of its crash."
[]
(let [key (queue-key)
k (ccfg/concurrency)]
(doseq [i (range k)]
(let [conn (connect!)]
(swap! pollers assoc i conn)
(l/info :hint "poller started" :id i :key key)
(poll-loop! conn key process!)))))
(defn stop!
"Ends the pollers, closing their connections. The job that got orphan
in a settle goes one row further to the next poller; no state hides
in this process."
[]
(doseq [conn (vals @pollers)]
(.quit ^js conn))
(reset! pollers {}))

9
exporter/test/exporter_tests/runner.cljs
View File

@ -18,14 +18,23 @@
[exporter-tests.scheduler-test]
[exporter-tests.shell-test]
[exporter-tests.wasm-pool-test]
[exporter-tests.worker-test]
[goog.object :as gobj]))
(enable-console-print!)
;; The tests that ride the mock http must never leave a rejected promise
;; dangling: a dangling one kills the whole run with ERR_UNHANDLED_REJECTION
;; and without any way to point at the chain that broke. Log and keep going.
(js/process.on "unhandledRejection"
(fn [reason _promise]
(println (str "UNHANDLED-REJECTION: " reason))))
(def test-namespaces
['exporter-tests.export-shapes-test
'exporter-tests.jobs-test
'exporter-tests.api-test
'exporter-tests.worker-test
'exporter-tests.consumer-config-test
'exporter-tests.renderer-svg-test
'exporter-tests.scheduler-test

254
exporter/test/exporter_tests/worker_test.cljs Normal file
View File

@ -0,0 +1,254 @@
;; 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 SUBSIDIARY SL
(ns exporter-tests.worker-test
"The queue loop of the worker: the payload it reads, the claim it
makes, and one settled export end to end, over a fake redis queue and
a fake management http. The e2e against the real backend is Task 8's
business in devenv; this harness pins the shapes the worker owns."
(:require
["node:fs/promises" :as fsp]
["node:path" :as path]
["undici" :as http]
[app.common.transit :as transit]
[app.common.uuid :as uuid]
[app.consumer.api :as api]
[app.consumer.exports :as exports]
[app.consumer.worker :as worker]
[app.renderer :as rd]
[app.util.shell :as sh]
[cljs.test :as t :include-macros true]
[cuerdas.core :as str]
[promesa.core :as p]))
;; ---- THE MANAGEMENT FAKE
(def ^:private original-fetch api/fetch)
(def ^:private calls
"One entry per management call: [method, body]."
(atom []))
(defn- method-of
"The management method the client asked for."
[uri]
(last (str/split (str uri) "/")))
(defn- steps-of
"The sequence of management methods the run went by."
[]
(mapv first @calls))
(defn- call-of
"The body of the first call of one method."
[method]
(some (fn [[m body]]
(when (= m method) body))
@calls))
(defn- fake-fetch!
"Installs a management http fake: `respond` answers with the `:action`
envelope per [method, body] (the multipart arrives as a FormData
instance). Every call lands in `calls` before the answer. Returns the
restore thunk."
[respond]
(reset! calls [])
(let [original api/fetch]
(set! api/fetch
(fn [uri request]
(let [method (method-of uri)
body (if (instance? http/FormData (.-body request))
(.-body request)
(transit/decode-str (.-body request)))]
(swap! calls conj [method body])
(p/resolved #js {:status 200
:text (constantly
(p/resolved (transit/encode-str
(respond method body))))}))))
(fn [] (set! api/fetch original))))
;; ---- THE RD FAKE
(def ^:private original-render rd/render)
(defn- fake-render!
"Installs a fake renderer that writes one file of `content` per
object and calls back with it. Returns the restore thunk."
[content]
(let [original rd/render]
(set! rd/render
(fn [_params on-object]
(p/let [rendered-path (sh/tempfile :prefix "penpot.render."
:suffix ".png")
_ (fsp/writeFile rendered-path content)]
(on-object {:path rendered-path
:filename "rendered.png"})
(p/resolved nil))))
(fn [] (set! rd/render original))))
(defn- fake-render-fail!
"A renderer that always fails: the failure path of the run."
[]
(let [original rd/render]
(set! rd/render (fn [_params _on-object]
(p/rejected (ex-info "render boom" {}))))
(fn [] (set! rd/render original))))
;; ---- THE QUEUE FAKE
(defn- fake-conn!
"A fake ioredis connection: `blpop` pops one payload from the queue
the atom holds, or nil."
[queue]
#js {:blpop (fn [_key _timeout-s]
(if-let [payload (first @queue)]
(do
(swap! queue rest)
(p/resolved #js ["the.queue.key" payload]))
(p/resolved nil)))})
;; ---- THE FIXTURE
(defn- export-params
"Params of a one-shape png job, in the job-def shape, the way the
claim carries them: the type as text."
[]
{:exports [{:file-id (str (uuid/next))
:page-id (str (uuid/next))
:object-id (str (uuid/next))
:type "png"
:name "test shape"
:suffix ""
:scale 1}]
:name "the export"})
(defn- encoded-claim
"The JSON payload the dispatcher pushes."
[job-id scheduled-at]
(js/JSON.stringify #js [(str job-id) scheduled-at]))
(defn- answer-with
"The management answers of a run: the claim gives the run away to
this poller with the params of the job, the session mints a real
one, every report and settle answers run. The `stages` atom records
every report's stage (they travel wrapped in `:progress`)."
[params sid stages]
(fn [method body]
(when (= method "report-job-progress")
(vswap! stages conj (get-in body [:progress :stage])))
(case method
"claim-job" {:action :run :name "export-assets" :params params}
"create-job-session" {:session-id sid :session-token "session-token"}
{:action :run})))
(defn- run-claim
"The claim payload of one call of `process!`."
[job-id]
{:job-id job-id
:scheduled-at "2026-10-07T09:00:00Z"})
(t/deftest read-decodes-the-dispatcher-payload
(t/testing "a payload in shape names the job and when it was pushed"
(let [job-id (uuid/next)]
(t/is (= {:job-id (str job-id)
:scheduled-at "2026-10-07T09:00:00Z"}
(worker/read-payload! (encoded-claim job-id
"2026-10-07T09:00:00Z"))))))
(t/testing "a corrupt payload is dropped, not thrown at the loop"
(t/is (nil? (worker/read-payload! "{not a payload")))))
(t/deftest poll-once-awaits-what-it-claims
(t/async done
(let [queue (atom [(encoded-claim (uuid/next) "2026-10-07T09:00:00Z")])
conn (fake-conn! queue)
handled (atom [])]
(p/let [_ (worker/poll-once! conn
"the.queue.key"
(fn [claim] (swap! handled conj claim)))]
(t/is (= 1 (count @handled)))
(done)))))
(t/deftest poll-once-drops-empty-pops-and-corrupt-payloads
(t/async done
(let [queue (atom ["{corrupt"])
conn (fake-conn! queue)]
(p/let [_ (worker/poll-once! conn
"the.queue.key"
(fn [_claim]
(t/is false "corrupt must not be handled")))
_ (worker/poll-once! conn
"the.queue.key"
(fn [_claim]
(t/is false "a corrupt pop has nothing to handle")))]
(t/is (empty? @queue))
(done)))))
(t/deftest process-drops-a-claim-the-worker-never-won
(t/async done
(let [respond! (fake-fetch!
(fn [method _body]
(case method
"claim-job" {:action :skip :status "cancelled"}
{:action :run})))
render! (fake-render! "should never render")]
(p/let [settled (worker/process!
{:job-id (uuid/next)
:scheduled-at "2026-10-07T09:00:00Z"})]
(t/testing "only the claim traveled: no session, no report, no render"
(t/is (= ["claim-job"] (steps-of)))
(t/is (nil? settled)))
(respond!)
(render!)
(done)))))
(t/deftest process-runs-the-claimed-export-to-a-multipart-settle
(t/async done
(let [job-id (uuid/next)
sid (uuid/next)
params (export-params)
stages (volatile! [])
respond! (fake-fetch! (answer-with params sid stages))
restore! (fake-render! "rendered!")]
(p/let [_ (worker/process! (run-claim job-id))]
(t/testing "the run went by claim, session, beats, settle"
(t/is (= ["claim-job" "create-job-session"
"report-job-progress" "report-job-progress" "report-job-progress"
"complete-job"]
(steps-of))))
(t/testing "the settle carried the real artifact, named and typed"
(let [fd (call-of "complete-job")]
(t/is (instance? http/FormData fd))
(t/is (= (str job-id) (.get fd "job-id")))
(t/is (= (str sid) (.get fd "session-id")))
(t/is (= "image/png" (.get fd "mtype")))))
(t/testing "the beats painted the vocabulary in order"
(t/is (= [:preparing :rendering :packaging] @stages)))
(respond!)
(restore!)
(done)))))
(t/deftest process-settles-a-failed-export-with-fail-job
(t/async done
(let [job-id (uuid/next)
sid (uuid/next)
params (export-params)
stages (volatile! [])
respond! (fake-fetch! (answer-with params sid stages))
restore! (fake-render-fail!)]
(p/let [_ (worker/process! (run-claim job-id))]
(t/testing "the settle was a fail, not a complete"
(t/is (= "fail-job" (last (steps-of)))))
(t/testing "the failure is the shape the backend stores"
(let [fail (call-of "fail-job")]
(t/is (= :internal (get-in fail [:error :type])))
(t/is (= :export-failed (get-in fail [:error :code])))
(t/is (= "render boom" (get-in fail [:error :hint])))
(t/is (= sid (:session-id fail)))))
(respond!)
(restore!)
(done)))))