github/penpot
1
0
Fork 0
mirror of https://github.com/penpot/penpot.git synced 2026-10-29 22:10:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

♻️ Render as an injected service with validation at the edge

Browse Source 
The renderer becomes a system service (:exporter/renderer) whose
instance is the render fn with its config closed over; the consumer
renders through that dependency and never touches the facade. render
takes the task map as is, and only render-fn validates, so a bad
batch is rejected at the single entry before anything is leased. The
worker names the renderer by ref, service keys follow the domain,
and main boots render workers on their own thread.

AI-assisted-by: muse-spark-1.3-contributor
This commit is contained in:
Andrey Antukh 2026-10-08 21:51:53 +00:00
parent 395ab4eefa
commit be3f7d39fd
14 changed files with 275 additions and 220 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -39,21 +39,21 @@ on this side: the row of the job, its cancel and its result are the backend's.
- New code grows under `exporter.*`, in parallel to the legacy `app.*` tree: `exporter.main` will replace `app.core` once every piece is ported; do not extend `app.*`.
- Lifecycle is `exporter.utils.system`: async-aware Integrant-style init/halt (`init-key`/`halt-key` multimethods, `ref`s, `weavejester.dependency` ordering, `^:async` + `cljs.core/await`, no promesa chains); tests are `^:async deftest` with `await` directly in the body.
- First ported service is `exporter.browser` (whole `app.browser` at once): the running instance is the raw pool, the public API takes it first (`exec` fails fast on a nil pool), the browser factory is injected through config for hermetic tests, and the new tree carries no promesa.
- `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.
- `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`), a thread guard in `start` (a render worker thread runs this same bundle and boots the legacy `app.wasm.worker` engine instead of the system — without it a warmed pool worker would nest pools and pollers and never post `ready`), `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.pool` follows the backend `db/run!` shape: `run [cfg f & args]` resolves the pool from `cfg` (the `:exporter.wasm/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 [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]`.
- `exporter.renderer` is the malli-validated facade with one entry point, the `render-fn` closure over `cfg`: it validates the task map (`:exports :on-object :check-cancelled`) and dispatches it through `render [cfg task]`, which assumes a valid task and only partitions and joins. The `:exporter/renderer` service builds the closure from pools (refs) plus static config. The consumer renders through the injected fn and never requires the facade 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]`.
- `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.api` is the backend management client (claim/progress/session/settle, transit bodies, multipart artifact), a faithful `^:async` port with no promesa chains and the same `fetch` def seam for hermetic tests.
- `exporter.consumer.config` holds the consumer knobs (`concurrency` floored at 1, `queue-key`, `queue-name`): the runner, the worker and `exporter.main` read it, so no namespace under `exporter.*` depends on `app.consumer.*` anymore.
- `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.
- `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 injected `:renderer` → settle before the next pop. The worker config carries the queue knobs plus the `:renderer` by ref; no render statics nest inside it. `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/renderer` (both pools as `ref`s, static uris/flags) plus `:exporter.consumer/worker` (`:renderer` by ref, `:concurrency`, `:queue-key`). Service keys follow the domain (`:exporter.wasm/pool`, `:exporter.consumer/worker`), written `::alias/name` where an alias covers them. 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

59
exporter/src/exporter/consumer.cljs
View File

@ -15,7 +15,9 @@
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.
session `create-job-session` opened. The batch renders through the
injected render fn (`:exporter/renderer` once wired): this namespace
never calls the facade directly.
Cancellation arrives through the beats: a `skip` answer from
`report-job-progress` says the job can no longer hear the worker
@ -39,8 +41,7 @@
[cuerdas.core :as str]
[exporter.consumer.api :as api]
[exporter.consumer.plan :as plan]
[exporter.jobs :as jobs]
[exporter.renderer :as renderer]))
[exporter.jobs :as jobs]))
(def ^:private report-throttle-ms 250)
(def ^:private watchdog-interval-ms 1000)
@ -189,21 +190,20 @@
(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]
[render 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
(await (render {: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))
: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))
@ -214,7 +214,7 @@
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]
[render beat plan check-cancelled]
(let [{:keys [job-id resource prepared total counter-kind]} plan
failure (volatile! nil)
rendered (volatile! 0)
@ -233,10 +233,9 @@
"_"))
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))
(await (render {: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))
@ -256,7 +255,7 @@
(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]
[render beat plan check-cancelled]
(let [{:keys [job-id resource prepared total]} plan
file-id (-> prepared first :file-id)
paths (volatile! [])
@ -270,23 +269,22 @@
(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
(await (render {:exports (mapv #(assoc % :job-id job-id
:is-wasm (:is-wasm plan))
prepared)
:on-object on-object
:check-cancelled check-cancelled))
: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]
[render 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)))
(:single? plan) (run-single render beat plan check-cancelled)
(:frames? plan) (run-frames render beat plan check-cancelled)
:else (run-multiple render beat plan check-cancelled)))
;; ---- THE SETTLE
@ -324,10 +322,11 @@
(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]
session, plan, render batch, multipart complete. `render` is the
injected render fn (one task map per batch: `:exports`, `:on-object`,
`:check-cancelled`); the claim names the job and `params` carries
the frozen job params."
[render {:keys [job-id]} params]
(let [session (atom nil)
stop-watchdog (atom nil)
_ (jobs/register job-id)
@ -346,7 +345,7 @@
(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))]
resource (await (run-prepared render 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

11
exporter/src/exporter/consumer/worker.cljs
View File

@ -26,8 +26,9 @@
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."
`:connect` factory) and the `:renderer` it renders through (the
`:exporter/renderer` service, injected by ref). There are no promesa
chains in this namespace."
(:require
["ioredis" :as redis]
[app.common.logging :as l]
@ -70,7 +71,7 @@
: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)))
(await (consumer/run-export (:renderer cfg) {:job-id job-id} (:params answer)))
(l/info :hint "job settled" :job-id (str job-id))
nil))))
@ -140,10 +141,10 @@
(mapv (fn [conn] (.quit ^js conn)) @(:conns handle))))
nil))
(defmethod system/init-key :exporter.consumer/worker
(defmethod system/init-key ::consumer/worker
[_ cfg]
(start cfg))
(defmethod system/halt-key :exporter.consumer/worker
(defmethod system/halt-key ::consumer/worker
[_ handle]
(stop handle))

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

@ -9,38 +9,49 @@
`app.main`: a `system-config` map wired with refs, and `start`,
`stop` and `restart` managing the running system.
This namespace will replace `app.core` once every piece of `app.*`
has been ported over. Until then the shadow build still boots
`app.core`; switching the entry point comes last."
Render workers run this same bundle, so the thread decides what gets
booted: one legacy render worker (`app.wasm.worker`, engines stay
untouched), or the system of the main thread. Without the guard a
warmed pool worker would boot pools and pollers of its own and never
post its `ready` handshake."
(:require
["node:process" :as proc]
["node:worker_threads" :as wt]
[app.common.logging :as l]
[app.config :as cf]
;; Loaded for their init-key/halt-key methods, which is what makes
;; the pools and the queue consumer part of the system below.
[exporter.browser]
[app.wasm.worker :as wasm.worker]
[exporter.browser :as browser]
[exporter.consumer :as-alias consumer]
[exporter.consumer.config :as ccfg]
[exporter.consumer.worker]
[exporter.utils.system :as system]
[exporter.wasm :as-alias wasm]
[exporter.wasm.pool]))
(l/setup! {:exporter :info})
(def system-config
"The production wiring. Env-derived values are read here, at the
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)}}})
wiring layer; the components themselves take plain data. The render
service owns the render config (pools by ref plus the static
config); the worker only names the renderer it renders through."
{::browser/pool
{:max (ccfg/concurrency)}
::wasm/pool
{:max (ccfg/concurrency)}
:exporter/renderer
{::browser/pool (system/ref :exporter.browser/pool)
::wasm/pool (system/ref :exporter.wasm/pool)
:base-uri (cf/get-internal-uri)
:public-uri (cf/get :public-uri)
:svgo? (contains? cf/flags :exporter-svgo)}
::consumer/worker
{:renderer (system/ref :exporter/renderer)
:concurrency (ccfg/concurrency)
:queue-key (ccfg/queue-key)}})
;; The running system map, or nil when nothing is started.
;; Counterpart of the backend's `app.system/system` var.
@ -84,13 +95,20 @@
(.on proc/default "SIGINT" (fn [] (proc/exit 0))))))
(defn ^:async start
"Boot the production wiring. Resolves to `:started`."
"Boot the production wiring. Resolves to `:started`. On a render
worker thread boots the render worker instead: it runs this same
bundle and owns no pools or pollers."
[]
(install-process-handlers)
(l/info :msg "starting exporter"
:workers (ccfg/concurrency)
:version (:full cf/version))
(await (start-custom system-config)))
(if ^boolean wt/isMainThread
(do
(l/info :msg "starting exporter"
:workers (ccfg/concurrency)
:version (:full cf/version))
(await (start-custom system-config)))
(do
(wasm.worker/main)
:started)))
(defn ^:async stop
"Halt the running system, if any. Resolves to `:stopped`."

52
exporter/src/exporter/renderer.cljs
View File

@ -17,7 +17,7 @@
`cfg` is task-free infrastructure with process lifetime: the pool
services under their system keys (`:exporter.browser/pool`,
`:exporter.wasm.pool/pool`) plus the static render config
`:exporter.wasm/pool`) plus the static render config
(`:base-uri`, `:public-uri`, `:svgo?`) — a view over the running
system map that each domain reads with its own keys. The task
map holds everything per-task: the `exports` data plus the two
@ -28,10 +28,17 @@
driver takes the same `[cfg params on-object check-cancelled]` shape and
implements cancel internally: cooperative checkpoints on the
browser, checkpoints plus a mid-render abort of its own leased
worker on wasm."
worker on wasm.
The renderer is also a system service (`:exporter/renderer`): its
config value carries the pools as refs plus the static config, and
its running instance is the render fn with that config closed over,
which is what the consumer renders through — the consumer never
calls this namespace directly."
(:require
[app.common.schema :as sm]
[exporter.browser.scope :as bscope]
[exporter.utils.system :as system]
[exporter.wasm.scope :as scope]))
(def schema:type
@ -79,21 +86,36 @@
(defn ^:async render
"Renders every export, firing them all at once: the browser ones
through their own scope, the wasm ones through one leased
worker. The whole task is validated before anything is leased, so a
bad task never touches a pool. Rejects on the first failure."
[cfg & {:as task}]
(let [{:keys [exports on-object check-cancelled]} (check-task task)
cfg (assoc cfg
::on-object on-object
::check-cancelled check-cancelled)
wasm-exports (filterv wasm? exports)
browser-exports (remove wasm? exports)]
worker. Assumes a valid task — validation lives one level up, at
the service entry, so through here a bad task never touches a pool
because it never arrives. Rejects on the first failure."
[cfg {:keys [exports on-object check-cancelled]}]
(let [cfg (assoc cfg
::on-object on-object
::check-cancelled check-cancelled)
wexports (filter wasm? exports)
bexports (remove wasm? exports)]
(await (js/Promise.all
(cond-> []
(seq browser-exports)
(conj (bscope/run cfg browser-exports))
(seq wasm-exports)
(seq bexports)
(conj (bscope/run cfg bexports))
(seq wexports)
(conj (scope/run cfg
(fn [render-fn]
(js/Promise.all (mapv render-fn wasm-exports))))))))
(js/Promise.all (map render-fn wexports))))))))
nil))
(defn render-fn
"The render entry of one running system: validates the task map and
dispatches it through the facade with `cfg` closed over, resolving
nil once every export landed. The single boundary where a bad batch
is rejected before anything is leased — and the dependency the
consumer renders through, so the consumer never calls the facade
directly."
[cfg]
(fn [task]
(render cfg (check-task task))))
(defmethod system/init-key :exporter/renderer
[_ cfg]
(render-fn cfg))

13
exporter/src/exporter/wasm/pool.cljs
View File

@ -31,7 +31,8 @@
[app.common.data :as d]
[app.common.exceptions :as ex]
[app.common.logging :as l]
[exporter.utils.system :as system]))
[exporter.utils.system :as system]
[exporter.wasm :as-alias wasm]))
(l/set-level! :info)
@ -118,7 +119,7 @@
:destroy destroy-instance
:validate validate-instance}))
(defmethod system/init-key ::pool
(defmethod system/init-key ::wasm/pool
[_ cfg]
(let [opts (merge defaults (d/without-nils cfg))
max-workers (:max opts)
@ -141,7 +142,7 @@
(await (.clear ^js pool))
nil)
(defmethod system/halt-key ::pool
(defmethod system/halt-key ::wasm/pool
[_ instance]
(when-let [pool (:pool instance)]
(l/info :hint "finalizing render worker pool")
@ -155,7 +156,7 @@
(fn? (unchecked-get x "release"))))
(defn ^:async run
"Lends one worker: resolves the pool from `cfg` (the `::pool`
"Lends one worker: resolves the pool from `cfg` (the `::wasm/pool`
service instance, or a raw pool), acquires a worker, applies `f` to
the cfg carrying it under `::worker` plus any extra `args`, and
returns the worker — or destroys it when `f` fails, the way a failed
@ -163,8 +164,8 @@
worker may be aborted or mid-write, and a terminated worker cannot
come back."
[cfg f & args]
(let [cfg (if (pool? cfg) {::pool {:pool cfg}} cfg)
pool (:pool (::pool cfg))]
(let [cfg (if (pool? cfg) {::wasm/pool {:pool cfg}} cfg)
pool (:pool (::wasm/pool cfg))]
(when-not (pool? pool)
(throw (ex/error :type :assertion
:code :invalid-pool-cfg

2
exporter/src/exporter/wasm/scope.cljs
View File

@ -110,7 +110,7 @@
timeout must never degrade silently into the default silence budget."
[cfg params]
(let [worker (::pool/worker cfg)
timeout-ms (or (:timeout-ms (:exporter.wasm.pool/pool cfg))
timeout-ms (or (:timeout-ms (:exporter.wasm/pool cfg))
(throw (ex/error :type :assertion
:code :timeout-ms-missing
:hint "render-on needs :timeout-ms in cfg")))]

85
exporter/test/exporter_tests/consumer_test.cljs
View File

@ -19,8 +19,7 @@
[cuerdas.core :as cstr]
[exporter.consumer :as consumer]
[exporter.consumer.api :as api]
[exporter.jobs :as jobs]
[exporter.renderer :as renderer]))
[exporter.jobs :as jobs]))
;; ---- THE MANAGEMENT FAKE
@ -63,38 +62,31 @@
;; ---- 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."
"A render fn in place of the service: writes one file of `content`
per export and calls back with it. When `captured` is an atom, it
also keeps the last batch 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)))))
(fn [{:keys [exports on-object] :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 {:path path
:filename "rendered.png"})))))
exports)]
(-> (js/Promise.all (clj->js writes))
(.then (fn [_] (resolve nil)) reject))))))))
(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))))
(fn [& _]
(js/Promise.reject (ex-info "render boom" {}))))
;; ---- THE FIXTURE
@ -129,9 +121,9 @@
(case method
"create-job-session" {:session-id sid :session-token "session-token"}
{:action :run})))
restore-render (stub-render "rendered!")]
render (stub-render "rendered!")]
(try
(await (consumer/run-export {} {:job-id job-id} (export-params)))
(await (consumer/run-export render {: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"
@ -150,18 +142,17 @@
(catch :default cause
(t/is false (str "unexpected failure: " (ex-message cause))))
(finally
(restore-fetch)
(restore-render)))))
(restore-fetch)))))
(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)
render (stub-render "rendered!" captured)
job-id (uuid/next)
share-id (uuid/next)]
(try
(await (consumer/run-export
{}
render
{:job-id job-id}
{:exports [{:file-id (str (uuid/next))
:page-id (str (uuid/next))
@ -187,16 +178,15 @@
(catch :default cause
(t/is false (str "unexpected failure: " (ex-message cause))))
(finally
(restore-fetch)
(restore-render)))))
(restore-fetch)))))
(t/deftest ^:async run-packs-two-shapes-as-zip
(let [restore-fetch (fake-fetch (answer-with (uuid/next) (volatile! [])))
restore-render (stub-render "rendered!")
render (stub-render "rendered!")
job-id (uuid/next)]
(try
(await (consumer/run-export
{}
render
{:job-id job-id}
{:exports [{:file-id (str (uuid/next))
:page-id (str (uuid/next))
@ -220,15 +210,14 @@
(catch :default cause
(t/is false (str "unexpected failure: " (ex-message cause))))
(finally
(restore-fetch)
(restore-render)))))
(restore-fetch)))))
(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)
render (stub-render-fail)
job-id (uuid/next)]
(try
(await (consumer/run-export {} {:job-id job-id} (export-params)))
(await (consumer/run-export render {: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)))))
@ -240,8 +229,7 @@
(catch :default cause
(t/is false (str "unexpected failure: " (ex-message cause))))
(finally
(restore-fetch)
(restore-render)))))
(restore-fetch)))))
(t/deftest ^:async cancel-before-the-render-settles-cancelled
(let [restore-fetch (fake-fetch
@ -251,10 +239,10 @@
:session-token "session-token"}
"report-job-progress" {:action :skip}
{:action :run})))
restore-render (stub-render "should never render")
render (stub-render "should never render")
job-id (uuid/next)]
(try
(await (consumer/run-export {} {:job-id job-id} (export-params)))
(await (consumer/run-export render {: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")]
@ -262,5 +250,4 @@
(catch :default cause
(t/is false (str "unexpected failure: " (ex-message cause))))
(finally
(restore-fetch)
(restore-render)))))
(restore-fetch)))))

61
exporter/test/exporter_tests/consumer_worker_test.cljs
View File

@ -9,11 +9,12 @@
makes, and the handoff to the runner, over a fake redis connection
and a fake management http."
(:require
["node:fs/promises" :as fsp]
["undici" :as http]
[app.common.transit :as transit]
[app.common.uuid :as uuid]
[cljs.test :as t :include-macros true]
[cuerdas.core :as cstr]
[exporter.consumer :as consumer]
[exporter.consumer.api :as api]
[exporter.consumer.worker :as worker]))
@ -30,7 +31,9 @@
(set! api/fetch
(fn [uri request]
(let [method (last (cstr/split (str uri) "/"))
body (transit/decode-str (.-body request))]
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
@ -58,8 +61,7 @@
(defn- test-cfg
[extra]
(merge {:concurrency 1
:queue-key "the.queue.key"
:render {}}
:queue-key "the.queue.key"}
extra))
(t/deftest read-decodes-the-dispatcher-payload
@ -119,36 +121,51 @@
(finally
(restore-fetch)))))
(t/deftest ^:async process-hands-the-claim-to-the-runner
(t/deftest ^:async process-renders-the-claim-through-the-injected-renderer
;; the whole vertical slice over fakes: the claim, the runner, the
;; injected render fn and the multipart settle
(let [restore-fetch (fake-fetch
(fn [method _body]
(case method
"claim-job" {:action :run
:name "export-assets"
:params {:exports [] :name "x"}}
"claim-job" {:action :run
:name "export-assets"
: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"}}
"create-job-session" {:session-id (uuid/next)
:session-token "session-token"}
{: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"}]
render (fn [{:keys [exports on-object]}]
(reset! ran exports)
(js/Promise.
(fn [resolve reject]
(-> (.then (fsp/writeFile "/tmp/penpot-worker-render.png" "rendered!")
(fn [_]
(on-object {:path "/tmp/penpot-worker-render.png"
:filename "rendered.png"})))
(.then (fn [_] (resolve nil)) reject)))))]
(try
(let [job-id (uuid/next)]
(await (worker/process (test-cfg {:render render})
(await (worker/process (test-cfg {:renderer 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)))))
(t/testing "the one shape rendered with uuids and the job stamped"
(t/is (= 1 (count @ran)))
(let [export (first @ran)]
(t/is (uuid? (:file-id export)))
(t/is (= job-id (:job-id export)))))
(t/testing "the run settled as a complete"
(t/is (= "complete-job" (last (mapv first @calls))))))
(catch :default cause
(t/is false (str "unexpected failure: " (ex-message cause))))
(finally
(restore-fetch)
(restore-run)))))
(restore-fetch)))))
(t/deftest ^:async start-opens-one-connection-per-poller-and-stop-quits-them
(try

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

@ -41,15 +41,20 @@
(defn- fake-config
[calls]
{:exporter.browser/pool {:max 1
:create-browser (fn []
(swap! calls conj :create-browser)
(js/Promise.resolve (stub-browser calls)))}
:exporter.wasm.pool/pool {:max 1
:min 0
:create-worker (fn []
(swap! calls conj :create-worker)
(js/Promise.resolve #js {}))}})
{:exporter.browser/pool {:max 1
:create-browser (fn []
(swap! calls conj :create-browser)
(js/Promise.resolve (stub-browser calls)))}
:exporter.wasm/pool {:max 1
:min 0
:create-worker (fn []
(swap! calls conj :create-worker)
(js/Promise.resolve #js {}))}
:exporter/renderer {:exporter.browser/pool (system/ref :exporter.browser/pool)
:exporter.wasm/pool (system/ref :exporter.wasm/pool)
:base-uri "http://internal/"
:public-uri "http://public/"
:svgo? false}})
(defn- pool-of
[]
@ -57,7 +62,7 @@
(defn- wasm-pool-of
[]
(:exporter.wasm.pool/pool @main/system))
(:exporter.wasm/pool @main/system))
(t/deftest ^:async install-process-handlers-registers-handlers
(try
@ -90,15 +95,18 @@
(t/is (= {:max (ccfg/concurrency)}
(:exporter.browser/pool main/system-config)))
(t/is (= {:max (ccfg/concurrency)}
(:exporter.wasm.pool/pool main/system-config))))
(t/testing "the queue consumer renders through a view of the running pools"
(:exporter.wasm/pool main/system-config))))
(t/testing "the renderer owns the render config, pools by ref"
(let [renderer (:exporter/renderer main/system-config)]
(t/is (= {:exporter.browser/pool (system/ref :exporter.browser/pool)
:exporter.wasm/pool (system/ref :exporter.wasm/pool)}
(select-keys renderer
[:exporter.browser/pool :exporter.wasm/pool])))))
(t/testing "the queue consumer names the renderer it renders through"
(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]))))))
(t/is (= (system/ref :exporter/renderer) (:renderer worker))))))
;; NOTE: no test boots the production wiring: the wasm pool warms one
;; worker eagerly (min 1), and in this process the worker script would
@ -111,6 +119,8 @@
(let [calls (atom [])]
(t/is (= :started (await (main/start-custom (fake-config calls)))))
(t/is (some? (pool-of)))
(t/testing "the renderer boots as a render fn over the resolved pools"
(t/is (fn? (:exporter/renderer @main/system))))
(t/testing "the wasm pool rides the same lifecycle"
(t/is (some? (:pool (wasm-pool-of))))
(t/is (= 300000 (:timeout-ms (wasm-pool-of)))))

72
exporter/test/exporter_tests/renderer_test.cljs
View File

@ -96,7 +96,7 @@
:base-uri (cf/get-internal-uri)
:public-uri (cf/get :public-uri)
:svgo? false
:exporter.wasm.pool/pool {:pool (stub-wasm-pool calls worker) :timeout-ms 5000}}))
:exporter.wasm/pool {:pool (stub-wasm-pool calls worker) :timeout-ms 5000}}))
(defn- never-cancelled
[calls]
@ -121,10 +121,10 @@
(try
(let [calls (atom [])
seen (atom [])]
(await (renderer/render (test-cfg calls)
:exports [(test-params :png nil)]
:on-object (fn [object] (swap! seen conj object) nil)
:check-cancelled (never-cancelled calls)))
(await ((renderer/render-fn (test-cfg calls))
{:exports [(test-params :png nil)]
:on-object (fn [object] (swap! seen conj object) nil)
:check-cancelled (never-cancelled calls)}))
(t/is (= ["a"] (mapv :id @seen)))
(t/is (some #(and (vector? %) (= :goto (first %))) @calls))
(t/is (not-any? #(and (vector? %) (= :post (first %))) @calls)))
@ -136,10 +136,10 @@
(let [calls (atom [])
job-id (uuid/next)
seen (atom [])]
(await (renderer/render (test-cfg calls)
:exports [(test-params :png {:is-wasm true :job-id job-id})]
:on-object (fn [object] (swap! seen conj object) nil)
:check-cancelled (never-cancelled calls)))
(await ((renderer/render-fn (test-cfg calls))
{:exports [(test-params :png {:is-wasm true :job-id job-id})]
:on-object (fn [object] (swap! seen conj object) nil)
:check-cancelled (never-cancelled calls)}))
(t/is (= [{:id "w1" :path "/tmp/w1.png"}] @seen))
(t/is (some #{[:post "render"]} @calls))
(t/is (some #{:check-cancelled} @calls))
@ -151,10 +151,10 @@
(try
(let [calls (atom [])]
(try
(await (renderer/render (test-cfg calls)
:exports [(dissoc (test-params :png nil) :objects)]
:on-object (fn [_] nil)
:check-cancelled (never-cancelled calls)))
(await ((renderer/render-fn (test-cfg calls))
{:exports [(dissoc (test-params :png nil) :objects)]
:on-object (fn [_] nil)
:check-cancelled (never-cancelled calls)}))
(t/is false "render should have rejected params without objects")
(catch :default cause
(t/is (= :data-validation (-> cause ex-data :code))))))
@ -165,10 +165,10 @@
(try
(let [calls (atom [])]
(try
(await (renderer/render (test-cfg calls)
:exports [(test-params :png nil)]
:on-object :not-a-fn
:check-cancelled (never-cancelled calls)))
(await ((renderer/render-fn (test-cfg calls))
{:exports [(test-params :png nil)]
:on-object :not-a-fn
:check-cancelled (never-cancelled calls)}))
(t/is false "render should have rejected a non-fn on-object")
(catch :default cause
(t/is (= :data-validation (-> cause ex-data :code))))))
@ -181,12 +181,12 @@
job-id (uuid/next)
cfg (test-cfg calls)
seen (atom [])]
(t/is (nil? (await (renderer/render cfg
:exports [(test-params :png {:is-wasm true :job-id job-id})
(test-params :png {:is-wasm true :job-id job-id})
(test-params :png {:job-id job-id})]
:on-object (fn [object] (swap! seen conj object) nil)
:check-cancelled (never-cancelled calls)))))
(t/is (nil? (await ((renderer/render-fn cfg)
{:exports [(test-params :png {:is-wasm true :job-id job-id})
(test-params :png {:is-wasm true :job-id job-id})
(test-params :png {:job-id job-id})]
:on-object (fn [object] (swap! seen conj object) nil)
:check-cancelled (never-cancelled calls)}))))
(t/is (= 1 (count (filter #{:acquire} @calls))))
(t/is (= 1 (count (filter #{:release} @calls))))
(t/is (= 3 (count @seen))))
@ -198,10 +198,10 @@
(let [calls (atom [])
cfg (test-cfg calls)
seen (atom [])]
(await (renderer/render cfg
:exports [(test-params :png nil)]
:on-object (fn [object] (swap! seen conj object) nil)
:check-cancelled (never-cancelled calls)))
(await ((renderer/render-fn cfg)
{:exports [(test-params :png nil)]
:on-object (fn [object] (swap! seen conj object) nil)
:check-cancelled (never-cancelled calls)}))
(t/is (= ["a"] (mapv :id @seen)))
(t/is (not-any? #{:acquire} @calls)))
(catch :default cause
@ -213,11 +213,11 @@
job-id (uuid/next)
cfg (test-cfg calls)]
(try
(await (renderer/render cfg
:exports [(test-params :png {:is-wasm true :job-id job-id})
(dissoc (test-params :png nil) :objects)]
:on-object (fn [_] nil)
:check-cancelled (never-cancelled calls)))
(await ((renderer/render-fn cfg)
{:exports [(test-params :png {:is-wasm true :job-id job-id})
(dissoc (test-params :png nil) :objects)]
:on-object (fn [_] nil)
:check-cancelled (never-cancelled calls)}))
(t/is false "render should have rejected the bad batch")
(catch :default cause
(t/is (= :data-validation (-> cause ex-data :code)))
@ -230,10 +230,10 @@
(try
(let [calls (atom [])]
(try
(await (renderer/render (test-cfg calls)
:exports [(test-params :png nil)]
:on-objec (fn [_] nil)
:check-cancelled (never-cancelled calls)))
(await ((renderer/render-fn (test-cfg calls))
{:exports [(test-params :png nil)]
:on-objec (fn [_] nil)
:check-cancelled (never-cancelled calls)}))
(t/is false "render should have rejected the misspelled key")
(catch :default cause
(t/is (= :data-validation (-> cause ex-data :code)))

2
exporter/test/exporter_tests/renderer_wasm_test.cljs
View File

@ -54,7 +54,7 @@
(defn- test-cfg
[calls worker]
{:exporter.wasm.pool/pool {:pool (stub-pool calls worker) :timeout-ms 5000}})
{:exporter.wasm/pool {:pool (stub-pool calls worker) :timeout-ms 5000}})
(defn- test-params
[]

14
exporter/test/exporter_tests/wasm_pool_service_test.cljs
View File

@ -38,21 +38,21 @@
(defn- test-config
[calls behavior timeout-ms]
{:exporter.wasm.pool/pool {:max 1
:min 0
:timeout-ms timeout-ms
:create-worker (fake-create-worker calls behavior)}})
{:exporter.wasm/pool {:max 1
:min 0
:timeout-ms timeout-ms
:create-worker (fake-create-worker calls behavior)}})
(defn- pool-of
[sys]
(:pool (:exporter.wasm.pool/pool sys)))
(:pool (:exporter.wasm/pool sys)))
(t/deftest ^:async pool-starts-and-stops-through-the-system
(try
(let [calls (atom [])
sys (await (system/init (test-config calls silent-behavior 5000)))]
(t/is (some? (:exporter.wasm.pool/pool sys)))
(t/is (= 5000 (:timeout-ms (:exporter.wasm.pool/pool sys))))
(t/is (some? (:exporter.wasm/pool sys)))
(t/is (= 5000 (:timeout-ms (:exporter.wasm/pool sys))))
(t/is (nil? (await (system/halt sys)))))
(catch :default cause
(t/is false (str "unexpected failure: " (ex-message cause))))))

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

@ -60,7 +60,7 @@
([calls worker on-object cancelled?]
(test-cfg calls worker on-object cancelled? nil))
([calls worker on-object cancelled? watch-interval]
(cond-> {:exporter.wasm.pool/pool {:pool (stub-pool calls worker) :timeout-ms 5000}
(cond-> {:exporter.wasm/pool {:pool (stub-pool calls worker) :timeout-ms 5000}
:exporter.renderer/on-object on-object
:exporter.renderer/check-cancelled (stub-check calls cancelled?)}
watch-interval (assoc :exporter.wasm.scope/watch-interval watch-interval))))
@ -160,7 +160,7 @@
(try
(await (scope/render-on {:exporter.renderer/on-object (fn [_] nil)
:exporter.renderer/check-cancelled (fn [] nil)
:exporter.wasm.pool/pool {:pool nil :timeout-ms 50}}
:exporter.wasm/pool {:pool nil :timeout-ms 50}}
(test-params)))
(t/is false "render-on should have rejected a missing lease")
(catch :default cause
@ -174,7 +174,7 @@
(await (scope/render-on {:exporter.wasm.pool/worker #js {}
:exporter.renderer/on-object (fn [_] nil)
:exporter.renderer/check-cancelled (fn [] nil)
:exporter.wasm.pool/pool {:pool nil}}
:exporter.wasm/pool {:pool nil}}
(test-params)))
(t/is false "render-on should have rejected a missing timeout")
(catch :default cause
@ -188,7 +188,7 @@
(let [calls (atom [])
worker (stub-worker calls silent-behavior)
cfg (-> (test-cfg calls worker (fn [_] nil) (atom false))
(assoc-in [:exporter.wasm.pool/pool :timeout-ms] 50))]
(assoc-in [:exporter.wasm/pool :timeout-ms] 50))]
(try
(await (scope/run cfg (fn [render-fn] (render-fn (test-params)))))
(t/is false "run should have rejected a silent worker")