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

🎉 Port the queue consumer to the new exporter tree

Browse Source 
The runner (exporter.consumer), its plan, the local cancel registry
(exporter.jobs) and the queue poller (exporter.consumer.worker) now
live beside the renderer: same beats, watchdog and settle, rendering
through the facade with one check shared by units of work and
drivers. The worker rides the system as :exporter.consumer/worker,
so the process has something holding it. The settle also releases
the registry entry the legacy runner leaked.

AI-assisted-by: muse-spark-1.3-contributor
This commit is contained in:
Andrey Antukh 2026-10-08 18:41:08 +00:00
parent b94174f1d1
commit 4eebf51041
12 changed files with 1245 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
.serena/memories/exporter/core.md
View File

@ -42,13 +42,16 @@ on this side: the row of the job, its cancel and its result are the backend's.
- `exporter.main` mirrors the backend's `app.main`: a `system-config` def reading env at the wiring layer, a public `system` atom with the running map, `start`/`stop`/`restart` (`^:async`, resolving `:started`/`:stopped`/`:restarted`), `start-custom` as the test seam, idempotent `install-process-handlers` for the `uncaughtException`/SIGTERM/SIGINT wiring (runs on `start`, never on namespace load), `^:dev/before-load-async`/`^:dev/after-load-async` hooks bridging the promise lifecycle to shadow's `done` callbacks (a hook failure logs, never sticks the reload), and `main` as the entry point: shadow boots `exporter.main/main`, and the old `app.core` reload path (`stop` and the `:devtools` wiring) sits commented out until `app.*` is fully ported. The entry leaves info logging enabled for the new tree (`setup!` on the `exporter` logger, mirroring the old `{:app :info}`); without it the default level filters everything below fatal.
- Service config maps take plain (non-namespaced) param keys (`{:exporter.browser/pool {:max 2}}`): the top-level key already scopes the component, so namespacing the params only adds noise. Component keys and error reasons stay namespaced.
- `exporter.wasm.pool` follows the backend `db/run!` shape: `run [cfg f & args]` resolves the pool from `cfg` (the `:exporter.wasm.pool/pool` service instance, or a raw pool), acquires a worker, applies `f` to the cfg carrying it under `:exporter.wasm.pool/worker` plus any extra args, and returns the worker — or destroys it when `f` fails, like a rollback. Pure pool mechanics only (`init`/`halt`, `terminate`, `await-ready`); the render protocol lives one level up.
- `exporter.wasm.scope` owns the lease: `run [sys on-scope]` lends one worker for the whole batch (serializer, per-lease signal buffer, mid-render abort interval), and the public low-level `render-on [cfg params]` renders on the leased worker — asserting `:worker-not-leased`/`:timeout-ms-missing` (`:assertion`, never a silent default: a bare `::key` resolves to the CURRENT ns, never to an alias, so cross-ns keys are literals or `::alias/key`) — with the rest preconfigured and namespaced by setter (`:exporter.renderer/on-object` from the caller, `:exporter.wasm.scope/cancel-buffer` from the lease); the check is mandatory at the lease entry and runs here when configured.
- `exporter.renderer.browser` merges the bitmap/pdf/svg flows into one `render [ctx params on-object]` (branching by type, `on-object` always awaited, unknown types raise); `ctx` carries what the job params do not freeze (`:browser-pool`, `:base-uri`, `:public-uri`, `:svgo?`). Dead xml predicates from the old tree did not survive the merge.
- `exporter.browser.scope` owns the browser batch: `run [sys task]` fires every export through its own checkout, all at once, bounded by the pool. No lease — a browser batch shares nothing but the pool, so fan-out only. The facade validates once, partitions by backend and joins both scopes.
- `exporter.wasm.scope` owns the lease: `run [cfg on-scope]` lends one worker for the whole batch (serializer, per-lease signal buffer, mid-render abort interval), and the public low-level `render-on [cfg params]` renders on the leased worker — asserting `:worker-not-leased`/`:timeout-ms-missing` (`:assertion`, never a silent default: a bare `::key` resolves to the CURRENT ns, never to an alias, so cross-ns keys are literals or `::alias/key`) — with the rest preconfigured and namespaced by setter (`:exporter.renderer/on-object` from the caller, `:exporter.wasm.scope/cancel-buffer` from the lease); the check is mandatory at the lease entry and runs here when configured.
- `exporter.renderer.browser` merges the bitmap/pdf/svg flows into one `render [cfg params on-object check-cancelled]` (branching by type, `on-object` always awaited, unknown types raise); `cfg` carries what the job params do not freeze (`:exporter.browser/pool`, `:base-uri`, `:public-uri`, `:svgo?`). Dead xml predicates from the old tree did not survive the merge.
- `exporter.browser.scope` owns the browser batch: `run [cfg exports]` fires every export through its own checkout, all at once, bounded by the pool. No lease — a browser batch shares nothing but the pool, so fan-out only. The facade validates once, partitions by backend and joins both scopes.
- `exporter.renderer.wasm` is a thin single-shot over the scope lease. The scope fn survives one level down, as `exporter.wasm.scope/run`.
- `exporter.renderer` is the malli-validated facade with one entry point: `render [cfg & task]` takes a task map (`:exports :on-object :check-cancelled`) and fires the whole batch at once — the browser exports through `exporter.browser.scope`, the wasm ones through one leased worker — resolving nil once every export landed (objects travel through `on-object`). `cfg` is task-free infrastructure with process lifetime (pools, uris, flags); everything per-task rides in the validated task map. One closed schema validates data plus injections, so a bad task — or a misspelled key — never touches a pool; task validation raises `:data-validation`. Each driver takes the same `[cfg params on-object check-cancelled]` shape (`headless?` is now `wasm?`: the browser also renders headless, so the old name lied) and implements cancel internally: cooperative checkpoints on the browser, checkpoints plus a mid-render abort of its own leased worker on wasm. The lease lives one level down, in `exporter.wasm.scope/run`, which the facade calls with the task keys set once in `cfg`; the browser batch goes through `exporter.browser.scope/run [cfg exports]`.
- The runner stays backend-blind: plan → render batch → settle, with beats/watchdog/settle untouched. Its `on-object` calls the same injected check the drivers use. When `exporter.jobs` is ported, its only cancel surface toward renderers is one zero-arg check constructor per job; the old `cancel-fns`/terminate-registration apparatus does not survive (abort ownership moved to the lease holder).
- `exporter.main` wires both pools (`:exporter.browser/pool` and `:exporter.wasm.pool/pool`, each sized to the worker concurrency). No test boots the production wiring: the wasm pool warms one worker eagerly and the worker script in a test process would be the test bundle itself; the wiring is asserted as data and the lifecycle rides fake configs.
- `exporter.jobs` is the local cancel arm, smaller than the legacy `app.jobs`: one flag per claimed job (`register`/`mark-cancelled` returning true/nil/`cancelled?`/`release`) and one zero-arg check constructor per job (`check-cancelled`, raising `:job-cancelled`). No terminate callbacks and no shared-array signal travel through here — the abort belongs to whoever holds the render (the per-lease interval of `exporter.wasm.scope` terminates its own worker). The runner releases the registry entry on every settle; the legacy runner never did (one leaked entry per job).
- `exporter.consumer.plan` is the render plan (`prepare-exports`, `count-objects`, `sanitize-file-regex`), dispatching on `exporter.renderer/wasm?` instead of the legacy `headless?`.
- `exporter.consumer` is the runner: `run-export [cfg claim params]` keeps the legacy beats/watchdog/settle (first breath, render session, multipart complete or `fail-job`, temp release) but renders through the facade (`render [cfg & task]`) with one check from `exporter.jobs` shared between units of work and drivers. `cfg` is the render view; the claim names the job, `params` carries the frozen job params. The settle never rejects; only bugs of this process do.
- `exporter.consumer.worker` is the queue poller (`read-payload`, `process`, `poll-once`, `start`/`stop` with a running flag, no promesa): K connections over the `:exporter` queue, claim → `run-export` through the `:render` view → settle before the next pop. `stop` quits every connection and ends every loop (the legacy loops kept failing turns after the stop); boot cleans orphaned temp files first.
- `exporter.main` wires both pools (`:exporter.browser/pool` and `:exporter.wasm.pool/pool`, each sized to the worker concurrency) plus `:exporter.consumer/worker` (`:concurrency`, `:queue-key`, the `:render` view with both pools as `ref`s and the static uris/flags). No test boots the production wiring: the wasm pool warms one worker eagerly and the worker script in a test process would be the test bundle itself; the wiring is asserted as data and the lifecycle rides fake configs.
- Async test vars prove their chain only when really awaited: a test passing with 0 assertions means the chain broke upstream.
## Render details

371
exporter/src/exporter/consumer.cljs Normal file
View File

@ -0,0 +1,371 @@
;; 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.consumer
"The runner of the `:export-assets` job.
Runs one export end to end: the render plan of
`exporter.consumer.plan` (the name transducers and the partitioning
the export surfaces always kept), the pdf assembly of the frames
run, the beats shaped by the progress contract (`:preparing`,
`:rendering`, `:packaging`, with the `objects` and `pages` counters
of every partition) and one settle: a `complete-job` multipart call
that stores the artifact and closes the session the backend minted
in the same step. The render runs as the job's owner through the
session `create-job-session` opened.
Cancellation arrives through the beats: a `skip` answer from
`report-job-progress` says the job can no longer hear the worker
(cancelled, aborted, settled). Between units of work the runner
raises, with the same check the drivers run; inside a render the
lease holder aborts its own worker. The local registry entry is
released on every settle — unlike the legacy runner, which only
released the temp files and leaked one entry per job.
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. There are no promesa chains in this namespace."
(:require
[app.common.exceptions :as ex]
[app.common.logging :as l]
[app.common.uuid :as uuid]
[app.consumer.api :as api]
[app.handlers.resources :as rsc]
[app.jobs.utils :as job.utils]
[app.util.shell :as sh]
[cuerdas.core :as str]
[exporter.consumer.plan :as plan]
[exporter.jobs :as jobs]
[exporter.renderer :as renderer]))
(def ^:private report-throttle-ms 250)
(def ^:private watchdog-interval-ms 1000)
;; ---- PROGRESS AND CANCEL (the milestones of the contract)
(defn- start-beats
"The reporter of one run's beats.
The beat reports the stage and the counters and asks the backend to
go on; a `skip` answer means nobody is listening anymore — the job
was cancelled or settled elsewhere — and the export must stop: the
reporter fires `on-cancel` once (what marks the local flag the
checks read; the in-flight render aborts itself) and returns the
answer, so the runner sees it at the next unit of work and raises.
Between forced beats the reporter throttles the way the legacy export
did, since hundreds of objects would otherwise be hundreds of HTTP
calls for information nobody reads at that resolution. An errored
beat answers `run`: the export itself must not fail because the beat
could not fly, and the backend that is unreachable has the lease to
decide what happens to the row. The beat never rejects."
[job-id on-cancel]
(let [last-beat (atom 0)
last (atom nil)
fired (atom false)
check (fn [answer]
(when (= :skip (:action answer))
(when (compare-and-set! fired false true)
(l/info :hint "job cancelled by backend"
:job-id (str job-id))
(on-cancel)))
answer)
report (^:async fn [stage counters]
(try
(check (await (api/report-job-progress job-id {:stage stage
:counters counters})))
(catch :default cause
(l/warn :hint "beat failed to land"
:job-id (str job-id)
:cause cause)
{:action :run})))
beat (^:async fn [stage counters & {:keys [force?]}]
(let [now (js/Date.now)
now? (or force?
(>= (- now @last-beat) report-throttle-ms))]
(if now?
(do (reset! last-beat now)
(reset! last {:stage stage :counters counters})
(await (report stage counters)))
{:action :run :throttled? true})))]
{:beat beat :last last}))
;; ---- THE WATCHDOG
(defn- with-watchdog
"The forced beats of one run: one every `watchdog-interval-ms`, and
they repeat the last report the run made, so a cancellation that
lands mid-render (no object landing, no unit of work ending) is seen
within that interval. `stop` clears it once the run settles."
[beat last]
(let [timer (js/setInterval
#(beat (:stage @last) (:counters @last) :force? true)
watchdog-interval-ms)]
(fn stop [] (js/clearInterval timer))))
(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- ->uuid
"One id of the claim into what the renderer wants: the claim carries
the job row JSON, every id a string, while the render spec only
takes uuid objects. Values that already are uuids (tests, callers in
process) pass through."
[v]
(if (string? v) (uuid/uuid v) v))
(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; every id arrives as text and the render
spec only takes uuid objects. 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)
(string? (:file-id item)) (update :file-id ->uuid)
(string? (:page-id item)) (update :page-id ->uuid)
(string? (:object-id item)) (update :object-id ->uuid)
(string? (:share-id item)) (update :share-id ->uuid)
(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."
[job-id token {:keys [exports force-multiple name skip-children is-wasm]}]
(let [items (normalize-items exports)
frames? (boolean (every? frame-item? items))
prepared (plan/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?
:job-id (->uuid job-id)
:prepared prepared
:total (plan/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- check-beat
"A forced beat after a unit of work: a `skip` there is a cancelled
job, the same exception the legacy surfaces raised, so everything
below unwinds."
[answer]
(when (= :skip (:action answer))
(throw (ex/error :type :internal
:code :job-cancelled
:hint "export job was cancelled")))
answer)
(defn- ^:async run-single
"One render, one object: the artifact IS the object, moved into the
resource path the multipart settle will name."
[cfg beat plan check-cancelled]
(let [{:keys [job-id resource counter-kind]} plan
export (-> plan :prepared first)
object (atom nil)]
(job.utils/track job-id (:path resource))
(check-beat (await (beat :rendering (counter counter-kind 0 1) :force? true)))
(await (renderer/render cfg
:exports [(assoc export
:job-id job-id
:skip-children (:skip-children plan))]
:on-object (fn [obj]
(reset! object obj)
(job.utils/track job-id (:path obj))
nil)
:check-cancelled check-cancelled))
(await (sh/move (:path @object) (:path resource)))
(check-beat (await (beat :packaging (counter counter-kind 1 1) :force? true)))
resource))
(defn- ^:async run-multiple
"The multi-object render: every prepared export renders through the
one batch of the facade, 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."
[cfg beat plan check-cancelled]
(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]}]
(check-cancelled)
(job.utils/track job-id path)
(vswap! rendered inc)
(beat :rendering
(counter counter-kind @rendered total))
(rsc/add-to-zip zip path
(str/replace filename
plan/sanitize-file-regex
"_"))
nil)]
(check-beat (await (beat :rendering (counter counter-kind 0 total) :force? true)))
(await (renderer/render cfg
:exports (mapv #(assoc % :job-id job-id) prepared)
:on-object append
:check-cancelled check-cancelled))
(when-let [cause @failure]
(throw cause))
(await (rsc/close-zip zip))
(check-beat (await (beat :packaging (counter counter-kind total total) :force? true)))
resource))
(defn- ^:async join-pdf
"The pages render one file per page; `pdfunite` stitches one pdf out
of them all, the way the legacy export did."
[job-id file-id paths]
(let [path (job.utils/track job-id
(sh/tempfile :prefix (str/concat "penpot.pdfunite." file-id ".")
:suffix ".pdf"))]
(await (sh/run-cmd "pdfunite" (into [] (concat (vec paths) [path]))))
path))
(defn- ^:async run-frames
"The frames render: a file per page, joined into the pdf of the file
once every page has landed."
[cfg beat plan check-cancelled]
(let [{:keys [job-id resource prepared total]} plan
file-id (-> prepared first :file-id)
paths (volatile! [])
rendered (volatile! 0)
on-object (fn [{:keys [path]}]
(check-cancelled)
(job.utils/track job-id path)
(vswap! paths conj path)
(vswap! rendered inc)
(beat :rendering
(counter :pages @rendered total))
nil)]
(check-beat (await (beat :rendering (counter :pages 0 total) :force? true)))
(await (renderer/render cfg
:exports (mapv #(assoc % :job-id job-id
:is-wasm (:is-wasm plan))
prepared)
:on-object on-object
:check-cancelled check-cancelled))
(let [joined (await (join-pdf job-id file-id @paths))]
(await (sh/move joined (:path resource)))
(check-beat (await (beat :packaging (counter :pages total total) :force? true)))
resource)))
(defn- run-prepared
[cfg beat check-cancelled plan]
(cond
(:single? plan) (run-single cfg beat plan check-cancelled)
(:frames? plan) (run-frames cfg beat plan check-cancelled)
:else (run-multiple cfg beat plan check-cancelled)))
;; ---- 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 cancelled?
"Whether `cause` ends a run the user asked to end: a cancelled
export is a normal ending, reported and logged as one, never as a
failure with a trace."
[cause]
(= :job-cancelled (:code (ex-data cause))))
(defn- ^:async 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]
(await (job.utils/release job-id))
(jobs/release job-id)
(try
(await (api/fail-job job-id (failure-report cause)
:session-id (some-> @session :session-id)))
(catch :default settle-cause
(l/warn :hint "unable to settle the failed job"
:job-id job-id :cause settle-cause))))
(defn ^:async run-export
"Runs one claimed export to its settle: first breath, render
session, plan, render batch, multipart complete. `cfg` is the render
config (a view over the running system); the claim names the job and
`params` carries the frozen job params."
[cfg {:keys [job-id]} params]
(let [session (atom nil)
stop-watchdog (atom nil)
_ (jobs/register job-id)
stop-cmd (fn [] (jobs/mark-cancelled job-id))
{:keys [beat last]} (start-beats job-id stop-cmd)
check-cancelled (jobs/check-cancelled job-id)]
(try
(reset! stop-watchdog (with-watchdog beat last))
;; the first breath right after the claim, before the
;; session roundtrip: the widget moves from queued to
;; exporting on it, and a cancel that landed meanwhile
;; raises here instead of after wasted work
(check-beat (await (beat :preparing {} :force? true)))
(let [session' (await (api/create-job-session job-id))]
(reset! session session')
(l/info :hint "render session minted"
:job-id (str job-id))
(let [plan (make-plan job-id (:session-token session') params)
resource (await (run-prepared cfg beat check-cancelled plan))]
;; the watchdog outlives the render, so the run stops it on
;; the way out: a run that ends badly stops beating too, or
;; its interval would knock on a dead job forever, one HTTP
;; call a second that only ever answers skip
(@stop-watchdog)
(let [artifact {:path (str (:path resource))
:filename (:filename resource)
:mtype (:mtype resource)}]
(await (api/complete-job-with-artifact
{:job-id job-id :session-id (:session-id session')}
artifact))
(await (job.utils/release job-id))
(jobs/release job-id)
(l/info :hint "export job settled" :job-id (str job-id)
:outcome "completed")
nil)))
(catch :default cause
(when-let [stop @stop-watchdog] (stop))
(if (cancelled? cause)
(l/info :hint "export job cancelled" :job-id (str job-id))
(l/error :hint "export job failed" :job-id (str job-id) :cause cause))
(await (settle-failure job-id session cause))))))

91
exporter/src/exporter/consumer/plan.cljs Normal file
View File

@ -0,0 +1,91 @@
;; 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.consumer.plan
"The render plan of an assets run.
The items the backend froze in the job become the renders this
process runs: file names built once and kept free of duplicates, then
grouped by scale and type, one render per group on the wasm engine
and one DOM page per partition of fifty on the browser engine.
`exporter.consumer` runs whatever this namespace plans, and this
namespace only plans."
(:require
[app.common.data :as d]
[app.util.mime :as mime]
[cuerdas.core :as str]
[exporter.renderer :as renderer]))
;; Regex to clean namefiles
(def sanitize-file-regex #"[\\/:*?\"<>|]")
(defn count-objects
[exports]
(reduce + 0 (map (comp count :objects) exports)))
(defn- assoc-file-name
"A transducer that assocs a candidate filename and avoid duplicates"
[]
(letfn [(find-candidate [params used]
(loop [index 0]
(let [candidate (str (:name params)
(:suffix params "")
(when (pos? index)
(str/concat "-" (inc index)))
(mime/get-extension (:type params)))]
(if (contains? used candidate)
(recur (inc index))
candidate))))]
(fn [rf]
(let [used (volatile! #{})]
(fn
([] (rf))
([result] (rf result))
([result params]
(let [candidate (find-candidate params @used)
params (assoc params :filename candidate)]
(vswap! used conj candidate)
(rf result params))))))))
(def ^:const ^:private
default-partition-size 50)
(defn prepare-exports
[exports token is-wasm]
(letfn [(process-group [[part1 :as group]]
;; The browser path renders one DOM page per partition; wasm
;; takes the whole group in one render.
(if (renderer/wasm? {:is-wasm is-wasm :type (:type part1)})
[(build-render group)]
(sequence (comp (partition-all default-partition-size)
(map build-render))
group)))
(build-render [[part1 :as part]]
(cond-> {:file-id (:file-id part1)
:page-id (:page-id part1)
:name (:name part1)
:token token
:type (:type part1)
:scale (:scale part1)
:objects (mapv part-entry->object part)}
;; a nil share-id is not an absent one for the render
;; spec: only name it when the item came through one
(some? (:share-id part1)) (assoc :share-id (:share-id part1))))
(part-entry->object [entry]
{:id (:object-id entry)
:filename (:filename entry)
:name (:name entry)
:suffix (:suffix entry)})]
(let [xform (comp
(map #(assoc % :token token))
(assoc-file-name))]
(->> (sequence xform exports)
(d/group-by (juxt :scale :type))
(map second)
(into [] (mapcat process-group))))))

149
exporter/src/exporter/consumer/worker.cljs Normal file
View File

@ -0,0 +1,149 @@
;; 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.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). Stopping quits
every connection and ends every loop: unlike the legacy worker,
whose loops kept failing turns after the stop, no poller survives
its handle.
The worker is a system component (`:exporter.consumer/worker`): its
config carries the queue knobs (`:concurrency`, `:queue-key`, the
`:connect` factory) and the `:render` view the runner renders
through. There are no promesa chains in this namespace."
(:require
["ioredis" :as redis]
[app.common.logging :as l]
[app.config :as cf]
[app.consumer.api :as api]
[app.consumer.config :as ccfg]
[app.jobs.utils :as job.utils]
[exporter.consumer :as consumer]
[exporter.utils.system :as system]))
(def ^:private poll-timeout-s 5)
(def ^:private reconnect-delay-ms 1000)
(defn- wait
[ms]
(js/Promise. (fn [resolve] (js/setTimeout resolve ms))))
(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 ^:async 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."
[cfg {:keys [job-id scheduled-at]}]
(let [answer (await (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))
(do
(l/info :hint "running job" :job-id (str job-id) :name (:name answer))
(await (consumer/run-export (:render cfg) {:job-id job-id} (:params answer)))
(l/info :hint "job settled" :job-id (str job-id))
nil))))
;; ---- THE POLLER
(defn ^:async 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]
(let [answer (await (.blpop ^js conn key poll-timeout-s))
payload (second answer)
claim (when (some? payload) (read-payload payload))]
(when (some? claim)
(await (on-payload claim))
nil)))
(defn- ^:async poll-loop
"The wait of one poller, turn after turn, until the handle stops. 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."
[handle conn key on-payload]
(try
(await (poll-once conn key on-payload))
(catch :default cause
(l/error :hint "poller turn failed" :key key :cause cause)
(await (wait reconnect-delay-ms))))
(when @(:running handle)
(await (poll-loop handle conn key on-payload))))
(defn- open-connection
"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 START
(defn ^:async start
"Launches the pollers over the queue and resolves the handle their
`stop` needs. A previous process leaves no temp files behind: the
boot cleans what the crash owned before the first pop."
[{:keys [concurrency queue-key connect] :as cfg}]
(await (job.utils/clean-orphans))
(let [running (atom true)
handle {:running running :conns (atom [])}
on-payload (fn [claim] (process cfg claim))
open (or connect open-connection)
k (or concurrency (ccfg/concurrency))]
(l/info :hint "worker started" :pollers k :queue queue-key)
(doseq [_ (range k)]
(let [conn (open)]
(swap! (:conns handle) conj conn)
(poll-loop handle conn queue-key on-payload)))
handle))
(defn ^:async stop
"Ends the pollers, closing their connections and their loops: the
running flag falls first, so a turn that ends after the quit does
not pop again. Idempotent: a second stop quits nothing."
[handle]
(when (and (some? handle)
(compare-and-set! (:running handle) true false))
(await (js/Promise.allSettled
(mapv (fn [conn] (.quit ^js conn)) @(:conns handle))))
nil))
(defmethod system/init-key :exporter.consumer/worker
[_ cfg]
(start cfg))
(defmethod system/halt-key :exporter.consumer/worker
[_ handle]
(stop handle))

69
exporter/src/exporter/jobs.cljs Normal file
View File

@ -0,0 +1,69 @@
;; 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.jobs
"The local arm of the job life this process runs.
The durable row of a job is the backend's business: the worker claims
it, beats against it and settles it through the management API. This
namespace keeps only the runtime bit that cannot live in a row — the
cancel flag the parts of a run read between renders — keyed by the
job id of the claimed job.
This is deliberately smaller than the legacy `app.jobs`: the abort
itself belongs to whoever holds the render (the per-lease interval of
`exporter.wasm.scope` terminates its own worker when the check
fails), so no terminate callbacks and no shared-array signal travel
through here. The only surface toward the renderers is one zero-arg
check constructor per job."
(:require
[app.common.exceptions :as ex]))
(defonce ^:private registry (atom {}))
(defn register
"The runtime record of a job this process runs: one cancel flag,
down until something marks it."
[job-id]
(swap! registry assoc (str job-id) {:cancelled? false})
nil)
(defn cancelled?
"True once the job has been marked cancelled in this process. The
check between objects reads it; inside a render the leased worker
reads its own abort instead."
[job-id]
(boolean (:cancelled? (get @registry (str job-id)))))
(defn mark-cancelled
"Signals one job of this process as cancelled: the flag the checks
read goes up. A job this process does not own, or a settled one, is
a no-op. Returns true when this call marked it, nil otherwise."
[job-id]
(let [k (str job-id)
rt (get @registry k)]
(when (some? rt)
(when-not (:cancelled? rt)
(swap! registry assoc-in [k :cancelled?] true)
true))))
(defn release
"Drops the local runtime entry once the job settled. The row stays in
the backend until it is read once more or expires."
[job-id]
(swap! registry dissoc (str job-id))
nil)
(defn check-cancelled
"The zero-arg check of one job: silent while it lives, raising
`:job-cancelled` once marked. The same fn travels into the render
task and runs between units of work."
[job-id]
(fn []
(when (cancelled? job-id)
(throw (ex/error :type :internal
:code :job-cancelled
:hint "export job was cancelled")))))

18
exporter/src/exporter/main.cljs
View File

@ -19,8 +19,9 @@
[app.config :as cf]
[app.consumer.config :as ccfg]
;; Loaded for their init-key/halt-key methods, which is what makes
;; the pools part of the system below.
;; the pools and the queue consumer part of the system below.
[exporter.browser]
[exporter.consumer.worker]
[exporter.utils.system :as system]
[exporter.wasm.pool]))
@ -28,9 +29,18 @@
(def system-config
"The production wiring. Env-derived values are read here, at the
wiring layer; the components themselves take plain data."
{:exporter.browser/pool {:max (ccfg/concurrency)}
:exporter.wasm.pool/pool {:max (ccfg/concurrency)}})
wiring layer; the components themselves take plain data. The worker
renders through a view of the running pools, resolved by refs, plus
the static render config."
{:exporter.browser/pool {:max (ccfg/concurrency)}
:exporter.wasm.pool/pool {:max (ccfg/concurrency)}
:exporter.consumer/worker {:concurrency (ccfg/concurrency)
:queue-key (ccfg/queue-key)
:render {:exporter.browser/pool (system/ref :exporter.browser/pool)
:exporter.wasm.pool/pool (system/ref :exporter.wasm.pool/pool)
:base-uri (cf/get-internal-uri)
:public-uri (cf/get :public-uri)
:svgo? (contains? cf/flags :exporter-svgo)}}})
;; The running system map, or nil when nothing is started.
;; Counterpart of the backend's `app.system/system` var.

266
exporter/test/exporter_tests/consumer_test.cljs Normal file
View File

@ -0,0 +1,266 @@
;; 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.consumer-test
"The export runner of the new tree: plan → render batch → settle,
over a fake management http and a stub renderer. The beats paint the
stage vocabulary in order and the settle carries the artifact; a
failure or a cancel settles as fail-job and never leaks the beat."
(:require
["node:fs/promises" :as fsp]
["undici" :as http]
[app.common.transit :as transit]
[app.common.uuid :as uuid]
[app.consumer.api :as api]
[app.util.shell :as sh]
[cljs.test :as t :include-macros true]
[cuerdas.core :as cstr]
[exporter.consumer :as consumer]
[exporter.jobs :as jobs]
[exporter.renderer :as renderer]))
;; ---- THE MANAGEMENT FAKE
(def ^:private calls
"One entry per management call: [method, body]."
(atom []))
(defn- method-of
[uri]
(last (cstr/split (str uri) "/")))
(defn- steps-of
[]
(mapv first @calls))
(defn- call-of
[method]
(some (fn [[m body]]
(when (= m method) body))
@calls))
(defn- fake-fetch
[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])
(js/Promise.resolve #js {:status 200
:text (constantly
(js/Promise.resolve
(transit/encode-str
(respond method body))))}))))
(fn [] (set! api/fetch original))))
;; ---- THE RENDER STUB
(defn- stub-render
"A renderer stub in place of the facade: writes one file of
`content` per object of every export and calls back with it. When
`captured` is an atom, it also keeps the last task it received."
([content] (stub-render content nil))
([content captured]
(let [original renderer/render]
(set! renderer/render
(fn [_cfg & {:as task}]
(when (some? captured)
(reset! captured task))
(js/Promise.
(fn [resolve reject]
(let [writes (mapv (fn [_]
(let [path (sh/tempfile :prefix "penpot.render."
:suffix ".png")]
(.then (fsp/writeFile path content)
(fn [_]
((:on-object task)
{:path path
:filename "rendered.png"})))))
(:exports task))]
(-> (js/Promise.all (clj->js writes))
(.then (fn [_] (resolve nil)) reject)))))))
(fn [] (set! renderer/render original)))))
(defn- stub-render-fail
[]
(let [original renderer/render]
(set! renderer/render
(fn [_cfg & _task]
(js/Promise.reject (ex-info "render boom" {}))))
(fn [] (set! renderer/render original))))
;; ---- THE FIXTURE
(defn- export-params
[]
{: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- answer-with
[sid stages]
(fn [method body]
(when (= method "report-job-progress")
(vswap! stages conj (get-in body [:progress :stage])))
(case method
"create-job-session" {:session-id sid :session-token "session-token"}
{:action :run})))
(t/deftest ^:async run-exports-one-shape-to-a-multipart-settle
(let [job-id (uuid/next)
sid (uuid/next)
stages (volatile! [])
restore-fetch (fake-fetch
(fn [method body]
(when (= method "report-job-progress")
(vswap! stages conj (get-in body [:progress :stage])))
(case method
"create-job-session" {:session-id sid :session-token "session-token"}
{:action :run})))
restore-render (stub-render "rendered!")]
(try
(await (consumer/run-export {} {:job-id job-id} (export-params)))
(t/testing "the run went by first breath, session, beats, settle"
(t/is (= ["report-job-progress" "create-job-session"
"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)))
(t/testing "the local registry entry is gone with the settle"
(t/is (false? (jobs/cancelled? job-id))))
(catch :default cause
(t/is false (str "unexpected failure: " (ex-message cause))))
(finally
(restore-fetch)
(restore-render)))))
(t/deftest ^:async run-hands-the-renderer-uuids-not-claim-strings
(let [restore-fetch (fake-fetch (answer-with (uuid/next) (volatile! [])))
captured (atom nil)
restore-render (stub-render "rendered!" captured)
job-id (uuid/next)
share-id (uuid/next)]
(try
(await (consumer/run-export
{}
{:job-id job-id}
{:exports [{:file-id (str (uuid/next))
:page-id (str (uuid/next))
:object-id (str (uuid/next))
:share-id (str share-id)
:type "png"
:name "test shape"
:suffix ""
:scale 1}]
:name "the export"}))
(let [export (-> @captured :exports first)]
(t/testing "every id arrived as a uuid object"
(t/is (uuid? (:file-id export)))
(t/is (uuid? (:page-id export)))
(t/is (uuid? (:share-id export)))
(t/is (= share-id (:share-id export)))
(t/is (uuid? (-> export :objects first :id))))
(t/testing "the job-id traveled with the render"
(t/is (= job-id (:job-id export))))
(t/testing "the task carried both injections"
(t/is (fn? (:on-object @captured)))
(t/is (fn? (:check-cancelled @captured)))))
(catch :default cause
(t/is false (str "unexpected failure: " (ex-message cause))))
(finally
(restore-fetch)
(restore-render)))))
(t/deftest ^:async run-packs-two-shapes-as-zip
(let [restore-fetch (fake-fetch (answer-with (uuid/next) (volatile! [])))
restore-render (stub-render "rendered!")
job-id (uuid/next)]
(try
(await (consumer/run-export
{}
{:job-id job-id}
{:exports [{:file-id (str (uuid/next))
:page-id (str (uuid/next))
:object-id (str (uuid/next))
:type "png"
:name "one"
:suffix ""
:scale 1}
{:file-id (str (uuid/next))
:page-id (str (uuid/next))
:object-id (str (uuid/next))
:type "png"
:name "two"
:suffix ""
:scale 1}]
:name "the export"}))
(t/testing "the artifact is the zip of both objects"
(let [fd (call-of "complete-job")]
(t/is (instance? http/FormData fd))
(t/is (= "application/zip" (.get fd "mtype")))))
(catch :default cause
(t/is false (str "unexpected failure: " (ex-message cause))))
(finally
(restore-fetch)
(restore-render)))))
(t/deftest ^:async a-failed-run-settles-fail-job-and-stops-beating
(let [restore-fetch (fake-fetch (answer-with (uuid/next) (volatile! [])))
restore-render (stub-render-fail)
job-id (uuid/next)]
(try
(await (consumer/run-export {} {:job-id job-id} (export-params)))
(await (js/Promise. (fn [resolve] (js/setTimeout resolve 1400))))
(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])))))
(catch :default cause
(t/is false (str "unexpected failure: " (ex-message cause))))
(finally
(restore-fetch)
(restore-render)))))
(t/deftest ^:async cancel-before-the-render-settles-cancelled
(let [restore-fetch (fake-fetch
(fn [method _body]
(case method
"create-job-session" {:session-id (uuid/next)
:session-token "session-token"}
"report-job-progress" {:action :skip}
{:action :run})))
restore-render (stub-render "should never render")
job-id (uuid/next)]
(try
(await (consumer/run-export {} {:job-id job-id} (export-params)))
(t/testing "the settle of a cancelled job was a fail the backend answers skip"
(t/is (= "fail-job" (last (steps-of))))
(let [fail (call-of "fail-job")]
(t/is (= :job-cancelled (get-in fail [:error :code])))))
(catch :default cause
(t/is false (str "unexpected failure: " (ex-message cause))))
(finally
(restore-fetch)
(restore-render)))))

173
exporter/test/exporter_tests/consumer_worker_test.cljs Normal file
View File

@ -0,0 +1,173 @@
;; 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.consumer-worker-test
"The queue loop of the new tree: the payload it reads, the claim it
makes, and the handoff to the runner, over a fake redis connection
and a fake management http."
(:require
[app.common.transit :as transit]
[app.common.uuid :as uuid]
[app.consumer.api :as api]
[cljs.test :as t :include-macros true]
[cuerdas.core :as cstr]
[exporter.consumer :as consumer]
[exporter.consumer.worker :as worker]))
;; ---- THE MANAGEMENT FAKE
(def ^:private calls
"One entry per management call: [method, body]."
(atom []))
(defn- fake-fetch
[respond]
(reset! calls [])
(let [original api/fetch]
(set! api/fetch
(fn [uri request]
(let [method (last (cstr/split (str uri) "/"))
body (transit/decode-str (.-body request))]
(swap! calls conj [method body])
(js/Promise.resolve #js {:status 200
:text (constantly
(js/Promise.resolve
(transit/encode-str
(respond method body))))}))))
(fn [] (set! api/fetch original))))
;; ---- THE QUEUE FAKE
(defn- fake-conn
[queue]
#js {:blpop (fn [_key _timeout-s]
(if-let [payload (first @queue)]
(do
(swap! queue rest)
(js/Promise.resolve #js ["the.queue.key" payload]))
(js/Promise.resolve nil)))
:quit (fn [] (js/Promise.resolve nil))})
(defn- encoded-claim
[job-id scheduled-at]
(js/JSON.stringify #js [(str job-id) scheduled-at]))
(defn- test-cfg
[extra]
(merge {:concurrency 1
:queue-key "the.queue.key"
:render {}}
extra))
(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 ^:async poll-once-awaits-what-it-claims
(try
(let [queue (atom [(encoded-claim (uuid/next) "2026-10-07T09:00:00Z")])
conn (fake-conn queue)
handled (atom [])]
(await (worker/poll-once conn
"the.queue.key"
(fn [claim]
(swap! handled conj claim)
(js/Promise.resolve nil))))
(t/is (= 1 (count @handled))))
(catch :default cause
(t/is false (str "unexpected failure: " (ex-message cause))))))
(t/deftest ^:async poll-once-drops-empty-pops-and-corrupt-payloads
(try
(let [queue (atom ["{corrupt"])
conn (fake-conn queue)]
(await (worker/poll-once conn
"the.queue.key"
(fn [_claim]
(t/is false "corrupt must not be handled"))))
(await (worker/poll-once conn
"the.queue.key"
(fn [_claim]
(t/is false "a corrupt pop has nothing to handle"))))
(t/is (empty? @queue)))
(catch :default cause
(t/is false (str "unexpected failure: " (ex-message cause))))))
(t/deftest ^:async process-drops-a-claim-the-worker-never-won
(let [restore-fetch (fake-fetch
(fn [method _body]
(case method
"claim-job" {:action :skip :status "cancelled"}
{:action :run})))]
(try
(await (worker/process (test-cfg {})
{:job-id (uuid/next)
:scheduled-at "2026-10-07T09:00:00Z"}))
(t/testing "only the claim traveled: no run, no render"
(t/is (= ["claim-job"] (mapv first @calls))))
(catch :default cause
(t/is false (str "unexpected failure: " (ex-message cause))))
(finally
(restore-fetch)))))
(t/deftest ^:async process-hands-the-claim-to-the-runner
(let [restore-fetch (fake-fetch
(fn [method _body]
(case method
"claim-job" {:action :run
:name "export-assets"
:params {:exports [] :name "x"}}
{:action :run})))
ran (atom nil)
restore-run (let [original consumer/run-export]
(set! consumer/run-export
(fn [cfg claim params]
(reset! ran [cfg claim params])
(js/Promise.resolve nil)))
(fn [] (set! consumer/run-export original)))
render {:some "render-cfg"}]
(try
(let [job-id (uuid/next)]
(await (worker/process (test-cfg {:render render})
{:job-id (str job-id)
:scheduled-at "2026-10-07T09:00:00Z"}))
(t/testing "the runner got the render view, the job and the params"
(t/is (= render (nth @ran 0)))
(t/is (= {:job-id (str job-id)} (nth @ran 1)))
(t/is (= {:exports [] :name "x"} (nth @ran 2)))))
(catch :default cause
(t/is false (str "unexpected failure: " (ex-message cause))))
(finally
(restore-fetch)
(restore-run)))))
(t/deftest ^:async start-opens-one-connection-per-poller-and-stop-quits-them
(try
(let [opened (atom [])
quit (atom [])
conn (fn []
(let [c (fake-conn (atom []))]
(swap! opened conj c)
(set! (.-quit c) (fn [] (swap! quit conj c) (js/Promise.resolve nil)))
;; the first pop never answers: the pollers sit in
;; it, so the test never races their loop
(set! (.-blpop c) (fn [_k _t] (js/Promise. (fn [_res _rej]))))
c))
handle (await (worker/start (test-cfg {:concurrency 2 :connect conn})))]
(t/is (= 2 (count @opened)))
(await (worker/stop handle))
(t/is (= 2 (count @quit)))
(t/testing "a second stop is a no-op"
(await (worker/stop handle))
(t/is (= 2 (count @quit)))))
(catch :default cause
(t/is false (str "unexpected failure: " (ex-message cause))))))

44
exporter/test/exporter_tests/exporter_consumer_plan_test.cljs Normal file
View File

@ -0,0 +1,44 @@
;; 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.exporter-consumer-plan-test
"The render plan of the new tree: same chunks and names as the
legacy planner, dispatching on the wasm flag."
(:require
[app.common.uuid :as uuid]
[cljs.test :as t :include-macros true]
[exporter.consumer.plan :as plan]))
(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 (plan/prepare-exports (exports 120 :png 1) "token" false)]
(t/is (= 3 (count parts)))
(t/is (= [50 50 20] (mapv (comp count :objects) parts)))))
(t/deftest wasm-exports-render-as-one-group
(let [parts (plan/prepare-exports (exports 120 :png 1) "token" true)]
(t/is (= 1 (count parts)))
(t/is (= 120 (count (:objects (first parts)))))))
(t/deftest duplicate-names-get-suffixed
(let [items [(assoc ((exports 1 :png 1) 0) :name "same")
(assoc ((exports 1 :png 1) 0) :name "same")]
parts (plan/prepare-exports items "token" false)]
(t/is (= ["same.png" "same-2.png"]
(mapv :filename (:objects (first parts)))))))

41
exporter/test/exporter_tests/exporter_jobs_test.cljs Normal file
View File

@ -0,0 +1,41 @@
;; 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.exporter-jobs-test
"The local cancel arm of the new tree: one flag per claimed job,
and the zero-arg check the renderers run."
(:require
[app.common.uuid :as uuid]
[cljs.test :as t :include-macros true]
[exporter.jobs :as jobs]))
(t/deftest register-mark-and-release
(let [job-id (uuid/next)]
(t/testing "a fresh job is not cancelled"
(jobs/register job-id)
(t/is (false? (jobs/cancelled? job-id))))
(t/testing "the first mark flips the flag and reports it"
(t/is (true? (jobs/mark-cancelled job-id)))
(t/is (true? (jobs/cancelled? job-id))))
(t/testing "a released job is gone, not cancelled"
(jobs/release job-id)
(t/is (false? (jobs/cancelled? job-id)))
(t/is (nil? (jobs/mark-cancelled job-id))))))
(t/deftest check-cancelled-raises-the-cooperative-code
(let [job-id (uuid/next)
check (jobs/check-cancelled job-id)]
(jobs/register job-id)
(t/testing "a live job passes the check silently"
(t/is (nil? (check))))
(t/testing "a marked job raises what the runner settles as cancel"
(jobs/mark-cancelled job-id)
(try
(check)
(t/is false "the check should have raised")
(catch :default cause
(t/is (= :job-cancelled (-> cause ex-data :code))))))
(jobs/release job-id)))

13
exporter/test/exporter_tests/main_test.cljs
View File

@ -13,7 +13,8 @@
[app.consumer.config :as ccfg]
[cljs.test :as t :include-macros true]
[exporter.browser :as browser]
[exporter.main :as main]))
[exporter.main :as main]
[exporter.utils.system :as system]))
(t/deftest new-tree-loads
(t/testing "exporter.main resolves once required"
@ -89,7 +90,15 @@
(t/is (= {:max (ccfg/concurrency)}
(:exporter.browser/pool main/system-config)))
(t/is (= {:max (ccfg/concurrency)}
(:exporter.wasm.pool/pool main/system-config)))))
(:exporter.wasm.pool/pool main/system-config))))
(t/testing "the queue consumer renders through a view of the running pools"
(let [worker (:exporter.consumer/worker main/system-config)]
(t/is (= (ccfg/concurrency) (:concurrency worker)))
(t/is (= (ccfg/queue-key) (:queue-key worker)))
(t/is (= {:exporter.browser/pool (system/ref :exporter.browser/pool)
:exporter.wasm.pool/pool (system/ref :exporter.wasm.pool/pool)}
(select-keys (:render worker)
[:exporter.browser/pool :exporter.wasm.pool/pool]))))))
;; NOTE: no test boots the production wiring: the wasm pool warms one
;; worker eagerly (min 1), and in this process the worker script would

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

@ -15,6 +15,10 @@
[exporter-tests.browser-test]
[exporter-tests.consumer-config-test]
[exporter-tests.consumer-plan-test]
[exporter-tests.consumer-test]
[exporter-tests.consumer-worker-test]
[exporter-tests.exporter-consumer-plan-test]
[exporter-tests.exporter-jobs-test]
[exporter-tests.jobs-test]
[exporter-tests.main-test]
[exporter-tests.renderer-browser-test]
@ -42,6 +46,10 @@
['exporter-tests.browser-scope-test
'exporter-tests.browser-test
'exporter-tests.consumer-plan-test
'exporter-tests.consumer-test
'exporter-tests.consumer-worker-test
'exporter-tests.exporter-consumer-plan-test
'exporter-tests.exporter-jobs-test
'exporter-tests.jobs-test
'exporter-tests.main-test
'exporter-tests.api-test