github/penpot
1
0
Fork 0
mirror of https://github.com/penpot/penpot.git synced 2026-09-01 17:49:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

🎉 Implement export jobs to process export requests (#11296)

Browse Source 
* ✨ Add export job model, store and scheduler to exporter

* ✨ Render wasm exports on pooled worker threads

* ✨ Add export job REST API to exporter

* ✨ Use export job API and allow cancelling wasm exports

* 🔧 Show export jobs in the internal debug panel

* 🔧 Pass flags and export job settings to the exporter container

* 📚 Document the exporter job API and its redis layout
This commit is contained in:
Elena Torró 2026-08-31 14:42:51 +02:00 committed by GitHub
parent ac5c88be66
commit 66b4a99ac3
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
40 changed files with 2934 additions and 706 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

55
backend/resources/app/templates/debug.tmpl
View File

@ -296,5 +296,60 @@ Debug Main Page
</form>
</fieldset>
</section>
</main>
<main class="dashboard wide">
<section class="widget wide">
<fieldset>
<legend>Export jobs:</legend>
<desc>
Export jobs as the exporter left them in redis. Records expire an hour
after the export settles, so this is a live view, not a history.
</desc>
<form method="get" action="/dbg">
<div class="row">
<input type="text" style="width:300px" name="job-id"
placeholder="filter by job id" value="{{export-job-filter}}" />
<input type="submit" value="Filter" />
<a href="/dbg">clear</a>
</div>
</form>
<div class="scroll-box">
<table>
<thead>
<tr>
<th>JOB ID</th>
<th>STATE</th>
<th>PROGRESS</th>
<th>CMD</th>
<th>BACKEND</th>
<th>NAME</th>
<th>CREATED</th>
<th>ENDED</th>
</tr>
</thead>
<tbody>
{% for job in export-jobs %}
<tr>
<td><tt>{{job.id}}</tt></td>
<td>{{job.state}}{% if job.interrupted %} (interrupted){% endif %}</td>
<td>{{job.done}} / {{job.total}}</td>
<td>{{job.cmd}}</td>
<td>{{job.backend}}</td>
<td>{{job.name}}</td>
<td>{{job.created-at}}</td>
<td>{{job.ended-at}}</td>
</tr>
{% empty %}
<tr><td colspan="8">No export jobs.</td></tr>
{% endfor %}
</tbody>
</table>
</div>
</fieldset>
</section>
</main>
{% endblock %}

29
backend/resources/app/templates/styles.css
View File

@ -143,6 +143,35 @@ nav > div:not(:last-child) {
height: fit-content;
}
/* A widget that holds a table rather than a form: full width, and tall
enough to be worth scrolling inside. */
.dashboard.wide {
margin-top: 0px;
}
.widget.wide {
max-width: none;
width: 100%;
}
.widget.wide .scroll-box {
max-height: 320px;
overflow-y: auto;
margin-top: 10px;
}
.widget.wide table {
width: 100%;
border-collapse: collapse;
}
.widget.wide th {
text-align: left;
position: sticky;
top: 0;
background: white;
}
.widget input[type=submit] {
outline: none;
border: 1px solid gray;

47
backend/src/app/http/debug.clj
View File

@ -26,6 +26,7 @@
[app.db :as db]
[app.features.file-migrations :as feat.fmig]
[app.http.session :as session]
[app.redis :as rds]
[app.rpc.commands.auth :as auth]
[app.rpc.commands.files-create :refer [create-file]]
[app.rpc.commands.profile :as profile]
@ -53,11 +54,53 @@
;; INDEX
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
(def ^:private max-export-jobs 200)
(defn- scan-export-job-keys
"Note: no index for now, get them all and filter"
[conn pattern]
(loop [cursor "0"
found []]
(let [[cursor keys] (rds/scan conn cursor pattern max-export-jobs)
found (into found keys)]
(if (or (nil? cursor)
(= "0" cursor)
(>= (count found) max-export-jobs))
(into [] (take max-export-jobs) found)
(recur cursor found)))))
(defn- get-export-jobs
[cfg job-id]
(let [filtered? (not (str/empty-or-nil? job-id))
job-uuid (when filtered? (parse-uuid job-id))]
(if (and filtered? (nil? job-uuid))
[]
(try
(let [pattern (str "penpot.exporter." (cf/get :tenant) ".job." (or job-uuid "*"))]
(->> (rds/run! cfg (fn [{:keys [::rds/conn]}]
(->> (scan-export-job-keys conn pattern)
(mapv (fn [key] (rds/hget conn key "data"))))))
(keep (fn [blob]
(try
(t/decode-str blob)
(catch Throwable _ nil))))
(sort-by :created-at #(compare %2 %1))
;; The exporter stores instants as epoch millis.
(map (fn [{:keys [created-at ended-at] :as job}]
(-> job
(assoc :created-at (some-> created-at ct/inst (ct/format-inst :rfc1123)))
(assoc :ended-at (some-> ended-at ct/inst (ct/format-inst :rfc1123))))))
(vec)))
(catch Throwable cause
(l/warn :hint "unable to read export jobs" :cause cause)
[])))))
(defn index-handler
[cfg request]
(let [profile-id (::session/profile-id request)
offset (clock/get-offset profile-id)
profile (profile/get-profile cfg profile-id)]
profile (profile/get-profile cfg profile-id)
job-filter (some-> request :params :job-id str/trim)]
{::yres/status 200
::yres/headers {"content-type" "text/html"}
::yres/body (-> (io/resource "app/templates/debug.tmpl")
@ -69,6 +112,8 @@
(ct/format-duration offset)
"NO OFFSET")
:current-time (ct/format-inst (ct/now) :http)
:export-jobs (get-export-jobs cfg job-filter)
:export-job-filter job-filter
:supported-features cfeat/supported-features}))}))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;

1
backend/src/app/main.clj
View File

@ -288,6 +288,7 @@
::http.debug/routes
{::db/pool (ig/ref ::db/pool)
::rds/pool (ig/ref ::rds/pool)
::session/manager (ig/ref ::session/manager)
::mbus/msgbus (ig/ref ::mbus/msgbus)
::sto/storage (ig/ref ::sto/storage)

39
backend/src/app/redis.clj
View File

@ -29,6 +29,7 @@
io.lettuce.core.api.sync.RedisScriptingCommands
io.lettuce.core.codec.RedisCodec
io.lettuce.core.codec.StringCodec
io.lettuce.core.KeyScanCursor
io.lettuce.core.KeyValue
io.lettuce.core.pubsub.api.sync.RedisPubSubCommands
io.lettuce.core.pubsub.RedisPubSubListener
@ -40,6 +41,8 @@
io.lettuce.core.RedisURI
io.lettuce.core.resource.ClientResources
io.lettuce.core.resource.DefaultClientResources
io.lettuce.core.ScanArgs
io.lettuce.core.ScanCursor
io.lettuce.core.ScriptOutputType
io.lettuce.core.SetArgs
io.netty.channel.nio.NioEventLoopGroup
@ -71,6 +74,8 @@
(-blpop [_ timeout keys])
(-eval [_ script])
(-get [_ key])
(-scan [_ cursor pattern limit])
(-hget [_ key field])
(-set [_ key val args])
(-del [_ key-or-keys])
(-ping [_]))
@ -205,6 +210,20 @@
(assert (string? key) "key expected to be string")
(.get cmd ^String key))
(-scan [_ cursor pattern limit]
(let [args (-> (ScanArgs.)
(.match ^String pattern)
(.limit (long limit)))
result (.scan cmd
^ScanCursor (ScanCursor/of ^String cursor)
^ScanArgs args)]
(MapEntry/create
(.getCursor ^KeyScanCursor result)
(vec (.getKeys ^KeyScanCursor result)))))
(-hget [_ key field]
(.hget cmd ^String key ^String field))
(-set [_ key val args]
(.set cmd
^String key
@ -345,6 +364,26 @@
(l/err :hint "timeout on get redis key" :key key :cause cause)
nil)))
(defn scan
[conn cursor pattern limit]
(assert (string? cursor) "cursor must be string instance")
(assert (string? pattern) "pattern must be string instance")
(try
(-scan conn cursor pattern limit)
(catch RedisCommandTimeoutException cause
(l/err :hint "timeout on scan" :pattern pattern :cause cause)
nil)))
(defn hget
[conn key field]
(assert (string? key) "key must be string instance")
(assert (string? field) "field must be string instance")
(try
(-hget conn key field)
(catch RedisCommandTimeoutException cause
(l/err :hint "timeout on hget" :key key :cause cause)
nil)))
(defn set
([conn key val]
(set conn key val nil))

14
docker/images/docker-compose.yaml
View File

@ -197,14 +197,24 @@ services:
- penpot
environment:
<< : [*penpot-secret-key, *penpot-public-uri]
<< : [*penpot-flags, *penpot-secret-key, *penpot-public-uri]
# Don't touch it; this uses an internal docker network to
# communicate with the frontend.
PENPOT_INTERNAL_URI: http://penpot-frontend:8080
## Valkey (or previously Redis) is used for the websockets notifications.
## Valkey (or previously Redis) is used for the websockets notifications
## and for storing the state export jobs
PENPOT_REDIS_URI: redis://penpot-valkey/0
# PENPOT_EXPORTER_MAX_CONCURRENT_JOBS: 4
# PENPOT_EXPORTER_MAX_JOBS_PER_PROFILE: 2
# PENPOT_EXPORTER_QUEUE_MAX: 64
# PENPOT_EXPORTER_JOB_TTL: 3600
# PENPOT_WASM_WORKER_POOL_MAX: 2
# PENPOT_WASM_WORKER_POOL_MIN: 1
# PENPOT_WASM_WORKER_IDLE_TIMEOUT: 300
# PENPOT_WASM_WORKER_IMAGE_CACHE_SIZE: 134217728
penpot-postgres:
image: "postgres:15"
restart: always

99
exporter/README.md Normal file
View File

@ -0,0 +1,99 @@
# Exporter
Node service that renders shapes and files to bitmap, SVG and PDF. Wasm exports
are **jobs**: created over HTTP, admitted by a scheduler with bounded
concurrency, and persisted in Redis so their state can be queried and cancelled.
The legacy entry point, which is what the browser backend still goes through,
runs the export as soon as it is asked for, with no admission control.
## HTTP API
Mounted under `/api/export` (the router matches on the path *after* that prefix,
so it also works when the process is hit directly on `/`).
| Method | Path | Description |
|----------|-----------------|----------------------------------------------------|
| `POST` | `/` | Legacy command multiplex; runs unscheduled |
| `POST` | `/jobs` | Create an export job |
| `GET` | `/jobs/{id}` | Job record |
| `DELETE` | `/jobs/{id}` | Request cancellation |
Job states: `queued` -> `running` -> `ended` | `error` | `cancelled`. The last
three are terminal.
## Redis layout
Every key is namespaced with `penpot.exporter.` plus the tenant
(`PENPOT_TENANT`, `default` in code but set to the workspace name in devenv,
e.g. `devenv-ws0`).
```
penpot.exporter.{tenant}.job.{job-id} hash field: data (transit blob of the
whole record)
penpot.exporter.{tenant}.job-cancel pubsub payload: the job id, one line
```
There is no index: the keyspace is one self-expiring hash per job and nothing
else. Each hash carries the same TTL as the exported file
(`PENPOT_EXPORTER_JOB_TTL`, default 3600s), refreshed on every write and never
after the job settles.
## Inspecting Redis
Redis is not published on the host, so `redis-cli` from your machine gets
connection refused. Run it **inside the devenv container**, against the `valkey`
host on database 0:
```bash
redis-cli -h valkey -n 0
```
`redis-cli -u "$PENPOT_REDIS_URI"` does the same and follows whatever the env is
set to (`redis://valkey/0` in devenv).
Keys carry the tenant, which in devenv is the **workspace name**
(`$PENPOT_TENANT`, e.g. `devenv-ws0`), not `default`. From the prompt:
```
# every job record
KEYS penpot.exporter.devenv-ws0.job.*
# the whole record, transit-json in the `data` field
HGET penpot.exporter.devenv-ws0.job.<job-id> data
# seconds left before the record expires
TTL penpot.exporter.devenv-ws0.job.<job-id>
# watch cancellations as they are published (blocks the connection)
SUBSCRIBE penpot.exporter.devenv-ws0.job-cancel
# drop one record
DEL penpot.exporter.devenv-ws0.job.<job-id>
```
`KEYS` is fine here -- the keyspace is a handful of job hashes. On a real
deployment use `SCAN 0 MATCH penpot.exporter.<tenant>.job.* COUNT 100` instead.
Do not `FLUSHDB`: the backend shares this database.
The backend debug UI also renders these records: `/dbg` has an *Export jobs*
section, with a `?job-id=` filter.
## Configuration
| Variable | Default | Description |
|---------------------------------------|---------|--------------------------------------|
| `PENPOT_REDIS_URI` | `redis://redis/0` | Job store and cancel topic |
| `PENPOT_TENANT` | `default` | Key and topic prefix |
| `PENPOT_EXPORTER_JOB_TTL` | `3600` | Lifetime of a job record, in seconds |
| `PENPOT_EXPORTER_MAX_CONCURRENT_JOBS` | `4` | Admission limit |
| `PENPOT_EXPORTER_MAX_JOBS_PER_PROFILE`| `2` | Per-profile admission limit |
| `PENPOT_EXPORTER_QUEUE_MAX` | `64` | Queue cap; over it, `429 :queue-full` |
| `PENPOT_WASM_WORKER_POOL_MAX` | `2` | Headless render worker threads; min 1 |
| `PENPOT_WASM_WORKER_POOL_MIN` | `1` | Workers kept warm; clamped to the max |
| `PENPOT_WASM_WORKER_IDLE_TIMEOUT` | `300` | Silence before a worker is terminated, in seconds |
| `PENPOT_WASM_WORKER_IMAGE_CACHE_SIZE` | `134217728` | Per-worker image cache budget, in bytes |
A headless job leases one render worker for its whole run, so it is admitted
only when a worker is free: `PENPOT_WASM_WORKER_POOL_MAX` is the real limit for
them, and `PENPOT_EXPORTER_MAX_CONCURRENT_JOBS` bounds the browser ones
alongside.

2
exporter/package.json
View File

@ -39,7 +39,7 @@
"check-fmt:clj": "cljfmt check --parallel=true src/ test/",
"lint:clj": "clj-kondo --parallel --lint src/ test/",
"build:test": "clojure -M:dev:shadow-cljs compile test",
"test": "pnpm run build:test && node target/tests/test.js",
"test": "pnpm run build:test && PENPOT_SECRET_KEY=${PENPOT_SECRET_KEY:-test-secret-key} node target/tests/test.js",
"test:quiet": "node ./scripts/test-quiet.js"
}
}

98
exporter/src/app/auth.cljs Normal file
View File

@ -0,0 +1,98 @@
;; This Source Code Form is subject to the terms of the Mozilla Public
;; License, v. 2.0. If a copy of the MPL was not distributed with this
;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
;;
;; Copyright (c) KALEIDOS INC Sucursal en España SL
(ns app.auth
"Resolves the caller's session cookie to a real profile id.
The export commands take `:profile-id` from the request body, which was
harmless while it only picked a pub/sub topic. The job API can read and
cancel other people's work, so its ownership comes from the session: the
token goes to the backend's `get-profile` command, which answers with the
anonymous profile (`uuid/zero`) when it is not a valid session.
Results are memoized briefly, so a burst of export calls from one client is
one round trip rather than one per request."
(:require
["undici" :as http]
[app.common.exceptions :as ex]
[app.common.logging :as l]
[app.common.transit :as t]
[app.common.uri :as u]
[app.common.uuid :as uuid]
[app.config :as cf]
[promesa.core :as p]))
(def ^:private cache-ttl-ms 60000)
(defonce ^:private cache (atom {}))
(defn- put-in-cache
"Stores the resolution by session token and removes expired entries.
Without cleanup, long-lived exporters accumulate stale entries"
[cache token profile-id now]
(-> (into {} (remove (fn [[_ {:keys [expires-at]}]] (<= expires-at now))) cache)
(assoc token {:profile-id profile-id
:expires-at (+ now cache-ttl-ms)})))
(defn- rpc-uri
[]
(-> (cf/get-internal-uri)
(u/ensure-path-slash)
(u/join "api/rpc/command/get-profile")
(str)))
(defn- fetch-profile-id
[token]
(let [uri (rpc-uri)
headers #js {"Content-Type" "application/transit+json"
"X-Shared-Key" (str "exporter " cf/management-key)
"Cookie" (str "auth-token=" token)}]
(->> (p/do (http/fetch uri #js {:method "POST" :headers headers :body (t/encode-str {})}))
(p/mcat (fn [^js resp]
(if (= 200 (.-status resp))
(.text resp)
(p/resolved nil))))
(p/fmap (fn [body]
(some-> body t/decode-str :id)))
(p/merr (fn [cause]
(l/warn :hint "unable to resolve session profile" :uri uri :cause cause)
(p/resolved nil))))))
(defn resolve-profile-id
"Promise of the authenticated profile id, or nil for an anonymous or absent
session."
[token]
(if (nil? token)
(p/resolved nil)
(let [{:keys [profile-id expires-at]} (get @cache token)]
(if (and expires-at (> expires-at (js/Date.now)))
(p/resolved profile-id)
(->> (fetch-profile-id token)
(p/fmap (fn [profile-id]
(let [profile-id (when (and profile-id (not= uuid/zero profile-id)) profile-id)]
(swap! cache put-in-cache token profile-id (js/Date.now))
profile-id))))))))
(defn require-profile-id
"Like `resolve-profile-id`, but rejects anonymous callers."
[token]
(->> (resolve-profile-id token)
(p/mcat (fn [profile-id]
(if profile-id
(p/resolved profile-id)
(ex/raise :type :authentication
:code :authentication-required
:hint "no valid session for this request"))))))
(defn check-owner!
"Raises unless `profile-id` owns `job`."
[job profile-id]
(when (or (nil? job)
(not= (str (:profile-id job)) (str profile-id)))
(ex/raise :type :not-found
:code :object-not-found
:hint "job does not exist"))
job)

20
exporter/src/app/config.cljs
View File

@ -28,7 +28,15 @@
:http-server-port 6061
:http-server-host "0.0.0.0"
:tempdir "/tmp/penpot"
:redis-uri "redis://redis/0"})
:redis-uri "redis://redis/0"
:exporter-max-concurrent-jobs 4
:exporter-max-jobs-per-profile 2
:exporter-queue-max 64
:exporter-job-ttl 3600
:wasm-worker-pool-max 2
:wasm-worker-pool-min 1
:wasm-worker-idle-timeout 300
:wasm-worker-image-cache-size (* 128 1024 1024)})
(def ^:private schema:config
[:map {:title "config"}
@ -42,7 +50,15 @@
[:redis-uri {:optional true} :string]
[:tempdir {:optional true} :string]
[:browser-pool-max {:optional true} ::sm/int]
[:browser-pool-min {:optional true} ::sm/int]])
[:browser-pool-min {:optional true} ::sm/int]
[:exporter-max-concurrent-jobs {:optional true} ::sm/int]
[:exporter-max-jobs-per-profile {:optional true} ::sm/int]
[:exporter-queue-max {:optional true} ::sm/int]
[:exporter-job-ttl {:optional true} ::sm/int]
[:wasm-worker-pool-max {:optional true} ::sm/int]
[:wasm-worker-pool-min {:optional true} ::sm/int]
[:wasm-worker-idle-timeout {:optional true} ::sm/int]
[:wasm-worker-image-cache-size {:optional true} ::sm/int]])
(def ^:private decode-config
(sm/decoder schema:config sm/string-transformer))

81
exporter/src/app/core.cljs
View File

@ -7,51 +7,88 @@
(ns app.core
(:require
["node:process" :as proc]
["node:worker_threads" :as wt]
[app.browser :as bwr]
[app.common.logging :as l]
[app.config :as cf]
[app.http :as http]
[app.jobs :as jobs]
[app.jobs.utils :as job.utils]
[app.redis :as redis]
[app.wasm :as wasm]
[app.wasm.pool :as wasm.pool]
[app.wasm.worker :as wasm.worker]
[promesa.core :as p]))
(enable-console-print!)
(l/setup! {:app :info})
(defn start
"Render workers run this same bundle, so the thread decides what gets booted:
the http server and its pools, or one render worker."
[& _]
(l/info :msg "initializing"
:public-uri (str (cf/get :public-uri))
:internal-uri (str (cf/get-internal-uri))
:version (:full cf/version))
(when (contains? cf/flags :wasm-export)
(l/warn :msg "headless wasm export enabled (experimental)"
:hint (str "renders run in-process on a single shared wasm module, "
"one at a time; not recommended for busy instances")
:wasm-dir wasm/artifact-dir
:image-cache-mb wasm/image-cache-mb))
(p/do!
(bwr/init)
(redis/init)
(http/init)))
(if-not ^boolean wt/isMainThread
(wasm.worker/main)
(do
(l/info :msg "initializing"
:public-uri (str (cf/get :public-uri))
:internal-uri (str (cf/get-internal-uri))
:version (:full cf/version))
(when (contains? cf/flags :wasm-export)
(l/info :msg "headless wasm export enabled (experimental)"
:wasm-dir wasm/artifact-dir
:workers (cf/get :wasm-worker-pool-max)
:image-cache-size (cf/get :wasm-worker-image-cache-size)))
(p/do
(bwr/init)
(redis/init)
(jobs/init)
(job.utils/init)
(wasm.pool/init)
(http/init)))))
(def main start)
;; Draining a pool waits for every checked-out resource to come back, which an
;; export in flight can hold for as long as its own timeout. On a hot reload
;; that would block `start` from ever running again, leaving a drained pool that
;; fails every later job.
(def ^:private shutdown-step-timeout 3000)
(defn- shutdown-step
[label f]
(-> (p/race [(p/do (f))
(p/fmap (constantly ::timeout) (p/delay shutdown-step-timeout))])
(p/handle (fn [result cause]
(when (or (some? cause) (= ::timeout result))
(l/warn :hint "shutdown step did not finish cleanly"
:step label
:cause cause))
nil))))
(defn stop
[done]
;; an empty line for visual feedback of restart
(js/console.log "")
(l/info :msg "stopping")
(p/do!
(bwr/stop)
(redis/stop)
(http/stop)
(done)))
(if-not ^boolean wt/isMainThread
;; A render worker owns no server, pools or connections; nothing to unwind.
(done)
(do
(l/info :msg "stopping")
(p/do
(shutdown-step "browser-pool" bwr/stop)
(shutdown-step "wasm-worker-pool" wasm.pool/stop)
(shutdown-step "redis" redis/stop)
(shutdown-step "http" http/stop)
(done)))))
(.on proc/default "uncaughtException"
(fn [cause]
(js/console.error cause)))
(.on proc/default "SIGTERM" (fn [] (proc/exit 0)))
(.on proc/default "SIGINT" (fn [] (proc/exit 0)))
;; Signals are only delivered to the main thread, and `exit` in a worker would
;; take down that worker rather than the process.
(when ^boolean wt/isMainThread
(.on proc/default "SIGTERM" (fn [] (proc/exit 0)))
(.on proc/default "SIGINT" (fn [] (proc/exit 0))))

74
exporter/src/app/handlers.cljs
View File

@ -6,22 +6,44 @@
(ns app.handlers
(:require
[app.auth :as auth]
[app.common.data :as d]
[app.common.exceptions :as ex]
[app.common.logging :as l]
[app.common.spec :as us]
[app.handlers.export-frames :as export-frames]
[app.handlers.export-shapes :as export-shapes]
[app.handlers.export :as export]
[app.util.transit :as t]
[clojure.spec.alpha :as s]
[cuerdas.core :as str]))
[promesa.core :as p]))
(l/set-level! :debug)
(def ^:private error-codes
#{:queue-full})
(defn on-error
[error exchange]
(let [{:keys [type code] :as data} (ex-data error)]
(cond
(and (= :validation type)
(contains? error-codes code))
(let [data {:type :validation
:code code
:hint (ex-message error)}]
(l/warn :hint "rejecting export request" :code code)
(-> exchange
(assoc :response/status 429)
(assoc :response/body (t/encode data))
(assoc :response/headers {"content-type" "application/transit+json"})))
(= :authentication type)
(let [data {:type :authentication
:code code
:hint (ex-message error)}]
(-> exchange
(assoc :response/status 401)
(assoc :response/body (t/encode data))
(assoc :response/headers {"content-type" "application/transit+json"})))
(or (= :validation type)
(= :assertion type))
(let [explain (us/pretty-explain data)
@ -62,27 +84,27 @@
(assoc :response/body (t/encode (d/without-nils data)))
(assoc :response/headers {"content-type" "application/transit+json"}))))))
(defmulti command-spec :cmd)
(s/def ::id ::us/string)
(s/def ::wait ::us/boolean)
(s/def ::cmd ::us/keyword)
(defmethod command-spec :export-shapes [_] ::export-shapes/params)
(defmethod command-spec :export-frames [_] ::export-frames/params)
(s/def ::params
(s/and (s/keys :req-un [::cmd]
:opt-un [::wait])
(s/multi-spec command-spec :cmd)))
(defn handler
[{:keys [:request/params] :as exchange}]
(let [{:keys [cmd] :as params} (us/conform ::params params)]
"The original `POST /api/export` entry point, and the one the browser backend
still goes through. The export runs as soon as it is asked for, and the
contract is unchanged: `:wait` answers with the finished resource, otherwise
with the resource handle while the work runs."
[{:keys [:request/params :request/auth-token] :as exchange}]
(let [{:keys [cmd wait] :as params} (export/conform-params params)]
(l/debug :hint "process-request" :cmd cmd)
(case cmd
:export-shapes (export-shapes/handler exchange params)
:export-frames (export-frames/handler exchange params)
(ex/raise :type :internal
:code :method-not-implemented
:hint (str/istr "method ~{cmd} not implemented")))))
(->> (auth/resolve-profile-id auth-token)
(p/mcat (fn [profile-id]
;; The session wins when there is one; the body value stays
;; the fallback so nothing that used to work stops working.
(export/export! auth-token (cond-> params
(some? profile-id)
(assoc :profile-id profile-id)))))
(p/mcat (fn [{:keys [resource pending]}]
(if wait
(p/fmap (fn [resource]
(assoc exchange :response/body resource))
pending)
(do
(p/merr (constantly nil) pending)
(p/resolved
(assoc exchange :response/body (dissoc resource :path))))))))))

103
exporter/src/app/handlers/export.cljs Normal file
View File

@ -0,0 +1,103 @@
;; This Source Code Form is subject to the terms of the Mozilla Public
;; License, v. 2.0. If a copy of the MPL was not distributed with this
;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
;;
;; Copyright (c) KALEIDOS INC Sucursal en España SL
(ns app.handlers.export
"Handle export jobs"
(:require
[app.common.spec :as us]
[app.handlers.export-frames :as export-frames]
[app.handlers.export-shapes :as export-shapes]
[app.jobs :as jobs]
[app.jobs.scheduler :as scheduler]
[app.jobs.utils :as job.utils]
[clojure.spec.alpha :as s]
[promesa.core :as p]))
;; --- PARAMS
(defmulti command-spec :cmd)
(s/def ::cmd ::us/keyword)
(s/def ::wait ::us/boolean)
(defmethod command-spec :export-shapes [_] ::export-shapes/params)
(defmethod command-spec :export-frames [_] ::export-frames/params)
(s/def ::params
(s/and (s/keys :req-un [::cmd]
:opt-un [::wait])
(s/multi-spec command-spec :cmd)))
(defn conform-params
[params]
(us/conform ::params params))
(defn- prepare
[cmd auth-token params]
(case cmd
:export-shapes (export-shapes/prepare auth-token params)
:export-frames (export-frames/prepare auth-token params)))
(defn- current
[job]
(or (jobs/lookup (:id job)) job))
(defn- run-and-track
[job run]
(->> (p/do (run job))
(p/mcat (fn [resource]
(->> (jobs/complete! (current job) resource)
(p/fmap (constantly resource)))))
(p/merr (fn [cause]
(if (jobs/cancelled? (:id job))
(p/rejected cause)
(->> (jobs/fail! (current job) cause)
(p/mcat (fn [_] (p/rejected cause)))))))))
(defn- run-now!
"Runs the job as soon as it is created, outside the scheduler."
[job]
(->> (p/do (jobs/start! job))
(p/mcat (fn [job] (p/do ((jobs/run-fn (:id job)) job))))
(p/fnly (fn [_ _]
(jobs/release! (:id job))
(job.utils/release! (:id job))))))
(defn- create!
[auth-token {:keys [cmd profile-id] :as params} start]
(let [{:keys [resource total headless run]} (prepare cmd auth-token params)]
(->> (jobs/create! {:profile-id profile-id
:cmd cmd
;; What the renderer will actually do, not what the
;; client asked for: `is-wasm` alone still renders in
;; the browser without the `wasm-export` flag, or for
;; svg, and the backend decides both the admission cap
;; and whether the client offers to cancel.
:backend (if headless "wasm" "browser")
:total total
:name (:name resource)
:resource-id (:id resource)}
(fn [job] (run-and-track job run)))
(p/fmap (fn [job]
(try
{:job job
:resource resource
:pending (start job)}
(catch :default cause
(jobs/fail! job cause)
(jobs/release! (:id job))
(throw cause))))))))
(defn create-job!
"Returns a promise of `{:job :resource :pending}`."
[auth-token params]
(create! auth-token params scheduler/submit!))
(defn export!
"Like `create-job!`, but the work starts right away: This is what
keeps browser exports behaving exactly as they did before there were jobs."
[auth-token params]
(create! auth-token params run-now!))

106
exporter/src/app/handlers/export_frames.cljs
View File

@ -6,19 +6,17 @@
(ns app.handlers.export-frames
(:require
[app.common.logging :as l]
[app.common.spec :as us]
[app.handlers.export-shapes :refer [prepare-exports]]
[app.handlers.export-shapes :refer [count-objects headless-exports? prepare-exports]]
[app.handlers.resources :as rsc]
[app.redis :as redis]
[app.jobs :as jobs]
[app.jobs.utils :as job.utils]
[app.renderer :as rd]
[app.util.shell :as sh]
[cljs.spec.alpha :as s]
[cuerdas.core :as str]
[promesa.core :as p]))
(declare ^:private handle-export)
(declare ^:private create-pdf)
(declare ^:private join-pdf)
(declare ^:private move-file)
@ -38,85 +36,53 @@
(s/keys :req-un [::exports]
:opt-un [::name ::is-wasm]))
(defn handler
[{:keys [:request/auth-token] :as exchange} {:keys [exports] :as params}]
;; NOTE: we need to have the `:type` prop because the exports
;; datastructure preparation uses it for creating the groups.
(let [exports (-> (map #(assoc % :type :pdf :scale 1 :suffix "") exports)
(prepare-exports auth-token))]
(handle-export exchange (assoc params :exports exports))))
(defn handle-export
[{:keys [:request/auth-token] :as exchange} {:keys [exports name profile-id is-wasm] :as params}]
(let [topic (str profile-id)
file-id (-> exports first :file-id)
resource
(rsc/create :pdf (or name (-> exports first :name)))
on-progress
(fn [done]
(let [data {:type :export-update
:resource-id (:id resource)
:status "running"
:done done}]
(redis/pub! topic data)))
on-complete
(fn [resource]
(let [data {:type :export-update
:resource-id (:id resource)
:resource-uri (:uri resource)
:name (:name resource)
:filename (:filename resource)
:mtype (:mtype resource)
:status "ended"}]
(redis/pub! topic data)))
on-error
(fn [cause]
(l/error :hint "unexpected error on frames exportation" :cause cause)
(let [data {:type :export-update
:resource-id (:id resource)
:name (:name resource)
:filename (:filename resource)
:status "error"
:cause (ex-message cause)}]
(redis/pub! topic data)))
result-cache
(atom [])
(defn- run-export
[job auth-token resource {:keys [exports is-wasm file-id]}]
(let [rendered (atom [])
on-object
(fn [{:keys [path] :as object}]
(let [res (swap! result-cache conj path)]
(on-progress (count res))))
(fn [{:keys [path] :as _object}]
(job.utils/track! (:id job) path)
(jobs/progress! job (count (swap! rendered conj path))))
procs
(->> (seq exports)
(map #(rd/render (assoc % :is-wasm is-wasm) on-object)))]
exports
(map #(assoc % :is-wasm is-wasm :job-id (:id job)) exports)]
(->> (p/all procs)
(p/fmap (fn [] @result-cache))
(p/mcat (partial join-pdf file-id))
(job.utils/track! (:id job) (:path resource))
(->> (rd/with-scope exports
(fn [render]
(jobs/check-cancelled! job)
(->> exports
(map (fn [export] (render export on-object)))
(p/all))))
(p/fmap (fn [_] @rendered))
(p/mcat (partial join-pdf job file-id))
(p/mcat (partial move-file resource))
(p/fmap (constantly resource))
(p/mcat (partial rsc/upload-resource auth-token))
(p/mcat (fn [resource]
(->> (sh/stat (:path resource))
(p/fmap #(merge resource %)))))
(p/merr on-error)
(p/fnly (fn [resource cause]
(when-not cause
(on-complete resource)))))
(p/fmap (fn [resource] (dissoc resource :path))))))
(assoc exchange :response/body (dissoc resource :path))))
(defn prepare
[auth-token {:keys [exports name is-wasm] :as _params}]
(let [exports (-> (map #(assoc % :type :pdf :scale 1 :suffix "") exports)
(prepare-exports auth-token is-wasm))
resource (rsc/create :pdf (or name (-> exports first :name)))
file-id (-> exports first :file-id)]
{:resource resource
:total (count-objects exports)
:headless (headless-exports? exports is-wasm)
:run (fn [job] (run-export job auth-token resource
{:exports exports
:is-wasm is-wasm
:file-id file-id}))}))
(defn- join-pdf
[file-id paths]
[job file-id paths]
(p/let [prefix (str/concat "penpot.pdfunite." file-id ".")
path (sh/tempfile :prefix prefix :suffix ".pdf")]
path (job.utils/track! (:id job) (sh/tempfile :prefix prefix :suffix ".pdf"))]
(apply sh/run-cmd! "pdfunite" (conj (vec paths) path))
path))

177
exporter/src/app/handlers/export_shapes.cljs
View File

@ -7,10 +7,10 @@
(ns app.handlers.export-shapes
(:require
[app.common.data :as d]
[app.common.logging :as l]
[app.common.spec :as us]
[app.handlers.resources :as rsc]
[app.redis :as redis]
[app.jobs :as jobs]
[app.jobs.utils :as job.utils]
[app.renderer :as rd]
[app.util.mime :as mime]
[app.util.shell :as sh]
@ -18,9 +18,6 @@
[cuerdas.core :as str]
[promesa.core :as p]))
(declare ^:private handle-single-export)
(declare ^:private handle-multiple-export)
(declare ^:private assoc-file-name)
(declare prepare-exports)
;; Regex to clean namefiles
@ -50,87 +47,92 @@
(s/keys :req-un [::exports ::profile-id]
:opt-un [::wait ::name ::skip-children ::force-multiple ::is-wasm]))
(defn handler
[{:keys [:request/auth-token] :as exchange} {:keys [exports force-multiple] :as params}]
(let [exports (prepare-exports exports auth-token)]
(if (and (not force-multiple)
(= 1 (count exports))
(= 1 (count (-> exports first :objects))))
(handle-single-export exchange (-> params
(assoc :export (first exports))
(dissoc :exports)))
(handle-multiple-export exchange (assoc params :exports exports)))))
(defn count-objects
[exports]
(reduce + 0 (map (comp count :objects) exports)))
(defn- handle-single-export
[{:keys [:request/auth-token] :as exchange} {:keys [export name skip-children is-wasm] :as params}]
(let [resource (rsc/create (:type export) (or name (:name export)))
export (assoc export :skip-children skip-children :is-wasm (boolean is-wasm))]
(defn- render!
[job export on-object]
(jobs/check-cancelled! job)
(rd/render (assoc export :job-id (:id job)) on-object))
(->> (rd/render export
(fn [{:keys [path] :as object}]
(sh/move! path (:path resource))))
(defn- scoped-renders
"Renders every export, the headless ones sharing a single worker."
[job exports on-object]
(rd/with-scope exports
(fn [render]
(jobs/check-cancelled! job)
(->> exports
(map (fn [export] (render export on-object)))
(p/all)))))
(defn- run-single
[job auth-token resource {:keys [export is-wasm skip-children]}]
(job.utils/track! (:id job) (:path resource))
(->> (render! job
(assoc export :skip-children skip-children :is-wasm (boolean is-wasm))
(fn [{:keys [path] :as _object}]
(job.utils/track! (:id job) path)
(sh/move! path (:path resource))))
(p/fmap (constantly resource))
(p/mcat (partial rsc/upload-resource auth-token))
(p/fmap (fn [resource] (dissoc resource :path)))))
(defn- run-multiple
[job auth-token resource {:keys [exports is-wasm]}]
(let [failure (volatile! nil)
zip (rsc/create-zip :resource resource
:on-error (fn [cause] (vreset! failure cause))
:on-progress (fn [{:keys [done]}]
(jobs/progress! job done)))
append (fn [{:keys [filename path] :as _object}]
(job.utils/track! (:id job) path)
(rsc/add-to-zip zip path (str/replace filename sanitize-file-regex "_")))]
(job.utils/track! (:id job) (:path resource))
(->> (scoped-renders job
(map #(assoc % :is-wasm (boolean is-wasm) :job-id (:id job)) exports)
append)
(p/mcat (fn [_]
(if-let [cause @failure]
(p/rejected cause)
(rsc/close-zip zip))))
(p/fmap (constantly resource))
(p/mcat (partial rsc/upload-resource auth-token))
(p/fmap (fn [resource]
(dissoc resource :path)))
(p/fmap (fn [resource]
(assoc exchange :response/body resource)))
(p/merr (fn [cause]
(l/error :hint "unexpected error on single export"
:cause cause)
(p/rejected cause))))))
(p/fmap (fn [resource] (dissoc resource :path))))))
(defn- handle-multiple-export
[{:keys [:request/auth-token] :as exchange} {:keys [exports wait profile-id name is-wasm] :as params}]
(let [resource (rsc/create :zip (or name (-> exports first :name)))
total (count exports)
topic (str profile-id)
(defn headless-exports?
"Whether any of `exports` renders headless, and so whether the job leases a
render worker. Mirrors what `rd/with-scope` decides at run time."
[exports is-wasm]
(boolean (some #(rd/headless? {:is-wasm is-wasm :type (:type %)}) exports)))
on-progress (fn [{:keys [done]}]
(when-not wait
(let [data {:type :export-update
:resource-id (:id resource)
:status "running"
:total total
:done done}]
(redis/pub! topic data))))
(defn prepare
[auth-token {:keys [exports force-multiple name skip-children is-wasm] :as _params}]
(let [exports (prepare-exports exports auth-token is-wasm)
headless? (headless-exports? exports is-wasm)
single? (and (not force-multiple)
(= 1 (count exports))
(= 1 (count (-> exports first :objects))))]
(if single?
(let [export (first exports)
resource (rsc/create (:type export) (or name (:name export)))]
{:resource resource
:total 1
:headless headless?
:run (fn [job] (run-single job auth-token resource
{:export export
:is-wasm is-wasm
:skip-children skip-children}))})
on-error (fn [cause]
(l/error :hint "unexpected error on multiple export" :cause cause)
(if wait
(p/rejected cause)
(redis/pub! topic {:type :export-update
:resource-id (:id resource)
:status "error"
:cause (ex-message cause)})))
zip (rsc/create-zip :resource resource
:on-error on-error
:on-progress on-progress)
append (fn [{:keys [filename path] :as resource}]
(rsc/add-to-zip zip path (str/replace filename sanitize-file-regex "_")))
proc (->> exports
(map (fn [export] (rd/render (assoc export :is-wasm (boolean is-wasm)) append)))
(p/all)
(p/mcat (fn [_] (rsc/close-zip zip)))
(p/fmap (constantly resource))
(p/mcat (partial rsc/upload-resource auth-token))
(p/fmap (fn [resource]
(let [data {:type :export-update
:name (:name resource)
:filename (:filename resource)
:resource-id (:id resource)
:resource-uri (:uri resource)
:mtype (:mtype resource)
:status "ended"}]
(p/do (redis/pub! topic data)
(assoc exchange :response/body resource)))))
(p/merr on-error))]
(if wait
(p/then proc #(assoc exchange :response/body (dissoc % :path)))
(assoc exchange :response/body (dissoc resource :path)))))
(let [resource (rsc/create :zip (or name (-> exports first :name)))]
{:resource resource
:total (count-objects exports)
:headless headless?
:run (fn [job] (run-multiple job auth-token resource
{:exports exports :is-wasm is-wasm}))}))))
(defn- assoc-file-name
"A transducer that assocs a candidate filename and avoid duplicates"
@ -160,13 +162,18 @@
default-partition-size 50)
(defn prepare-exports
[exports token]
(letfn [(process-group [group]
(sequence (comp (partition-all default-partition-size)
(map process-partition))
group))
[exports token is-wasm]
(letfn [(process-group [[part1 :as group]]
;; The browser renders a partition as a single DOM page, so it is
;; chunked to bound that page. A wasm export is headless, so
;; it does not need to be chunked, and can be rendered as a single partition.
(if (rd/headless? {:is-wasm is-wasm :type (:type part1)})
[(build-render group)]
(sequence (comp (partition-all default-partition-size)
(map build-render))
group)))
(process-partition [[part1 :as part]]
(build-render [[part1 :as part]]
{:file-id (:file-id part1)
:page-id (:page-id part1)
:share-id (:share-id part1)

60
exporter/src/app/handlers/jobs.cljs Normal file
View File

@ -0,0 +1,60 @@
;; This Source Code Form is subject to the terms of the Mozilla Public
;; License, v. 2.0. If a copy of the MPL was not distributed with this
;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
;;
;; Copyright (c) KALEIDOS INC Sucursal en España SL
(ns app.handlers.jobs
"REST surface for export jobs, under `/api/export/jobs`.
Ownership always comes from the session (see `app.auth`), never from the
request body."
(:require
[app.auth :as auth]
[app.common.exceptions :as ex]
[app.common.logging :as l]
[app.handlers.export :as export]
[app.jobs :as jobs]
[promesa.core :as p]))
(defn create
[{:keys [:request/auth-token :request/params] :as exchange}]
(->> (auth/require-profile-id auth-token)
(p/mcat (fn [profile-id]
(let [params (-> params
(assoc :profile-id profile-id)
(export/conform-params))]
(l/dbg :hint "create export job" :cmd (:cmd params) :profile-id (str profile-id))
(export/create-job! auth-token params))))
(p/fmap (fn [{:keys [job resource pending]}]
;; A failure is reported through the job record, so the
;; promise must not surface as an unhandled rejection.
(p/merr (constantly nil) pending)
(-> exchange
(assoc :response/body (-> (or (jobs/lookup (:id job)) job)
(assoc :filename (:filename resource))
(assoc :mtype (:mtype resource)))))))))
(defn fetch
[{:keys [:request/auth-token] :as exchange} job-id]
(->> (auth/require-profile-id auth-token)
(p/mcat (fn [profile-id]
(->> (jobs/fetch job-id)
(p/fmap #(auth/check-owner! % profile-id)))))
(p/fmap (fn [job]
(assoc exchange :response/body job)))))
(defn cancel
[{:keys [:request/auth-token] :as exchange} job-id]
(->> (auth/require-profile-id auth-token)
(p/mcat (fn [profile-id]
(->> (jobs/fetch job-id)
(p/fmap #(auth/check-owner! % profile-id)))))
(p/mcat (fn [job] (jobs/cancel! (:id job))))
(p/mcat (fn [_] (jobs/fetch job-id)))
(p/fmap (fn [job]
(if job
(assoc exchange :response/body job)
(ex/raise :type :not-found
:code :object-not-found
:hint "job does not exist"))))))

5
exporter/src/app/http.cljs
View File

@ -15,6 +15,7 @@
[app.common.transit :as t]
[app.config :as cf]
[app.handlers :as handlers]
[app.router :as router]
[cuerdas.core :as str]
[lambdaisland.uri :as u]
[promesa.core :as p]))
@ -94,7 +95,7 @@
size (js/Buffer.byteLength data "utf-8")]
(-> exchange
(assoc :response/body data)
(assoc :response/status 200)
(assoc :response/status (or status 200))
(update :response/headers assoc "content-type" "application/transit+json")
(update :response/headers assoc "content-length" size)))
@ -159,7 +160,7 @@
(defn init
[]
(let [handler (-> handlers/handler
(let [handler (-> (router/create handlers/handler)
(wrap-health)
(wrap-auth "auth-token")
(wrap-response-format)

284
exporter/src/app/jobs.cljs Normal file
View File

@ -0,0 +1,284 @@
;; This Source Code Form is subject to the terms of the Mozilla Public
;; License, v. 2.0. If a copy of the MPL was not distributed with this
;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
;;
;; Copyright (c) KALEIDOS INC Sucursal en España SL
(ns app.jobs
"Export job model and lifecycle.
The record is persisted in redis (`app.jobs.store`); the runtime bits that
cannot be serialized -- cancel callbacks, the cancel signal shared with a
render worker, the throttling bookkeeping -- stay in this process, keyed by
job id.
Every state change also publishes the same `:export-update` message the
exporter has always published, so websocket clients keep working unchanged."
(:require
[app.common.data :as d]
[app.common.exceptions :as ex]
[app.common.logging :as l]
[app.common.time :as ct]
[app.common.uuid :as uuid]
[app.jobs.store :as store]
[app.redis :as redis]
[promesa.core :as p]))
(l/set-level! :debug)
;; A large export reports progress per object; persisting each one would be
;; hundreds of writes for information nobody reads at that resolution.
(def ^:private progress-throttle-ms 250)
(def ^:private terminal-states #{"ended" "error" "cancelled"})
(defonce ^:private registry (atom {}))
(defn- now-ms
[]
(inst-ms (ct/now)))
(defn- runtime
[job-id]
(get @registry (str job-id)))
(defn lookup
"The live record of a job running in this process, or nil."
[job-id]
(:job (runtime job-id)))
(defn fetch
"The job record from the shared store."
[job-id]
(store/fetch job-id))
(defn- publish!
[{:keys [id profile-id resource-id state done total name filename mtype
resource-uri error]}]
(redis/pub! (redis/->tenant-key (str profile-id))
(d/without-nils
{:type :export-update
:job-id id
:resource-id resource-id
:status state
:done done
:total total
:name name
:filename filename
:mtype mtype
:resource-uri resource-uri
:cause error})))
(defn- store-job!
[job]
(swap! registry update (str (:id job)) assoc :job job)
job)
(defn create!
"Builds a queued job and persists it. `run-fn` is a 1-arg fn of the job that
performs the export and returns a promise; the caller decides when to run
it."
[{:keys [profile-id cmd backend total name resource-id]} run-fn]
(let [job {:id (uuid/next)
:profile-id profile-id
:cmd cmd
:backend backend
:state "queued"
:done 0
:total total
:name name
:resource-id resource-id
:created-at (now-ms)}]
(swap! registry assoc (str (:id job)) {:job job :run-fn run-fn :cancelled? false})
(->> (store/persist! job)
(p/fmap (constantly job)))))
(defn run-fn
[job-id]
(:run-fn (runtime job-id)))
(defn cancelled?
[job-id]
(boolean (:cancelled? (runtime job-id))))
(defn terminal?
[job]
(contains? terminal-states (:state job)))
(defn check-cancelled!
"Raises when the job has been cancelled. Called as each render's turn comes
up, so a cancellation stops the ones that have not started yet."
[job]
(when (cancelled? (:id job))
(ex/raise :type :internal
:code :job-cancelled
:hint "export job was cancelled")))
(defn cancel-signal
"Int32Array over a SharedArrayBuffer, readable from a worker thread: 0 while
the job is live, 1 once it has been cancelled. Nil once the job has been
released -- writing the signal back would leave an entry for a settled job in
the registry that nothing would ever remove."
[job-id]
(let [k (str job-id)]
(when-let [rt (get @registry k)]
(or (:cancel-signal rt)
(let [signal (js/Int32Array. (js/SharedArrayBuffer. 4))]
(swap! registry update k (fn [rt] (some-> rt (assoc :cancel-signal signal))))
(when (cancelled? job-id)
(js/Atomics.store signal 0 1))
signal)))))
(defn on-cancel
"Registers a callback used to abort the job's in-flight work (terminating a
render worker). A job fans out over several renders, so callbacks
accumulate. One registered for a job that already settled is dropped: keeping
it would revive that job's registry entry for good."
[job-id f]
(swap! registry update (str job-id)
(fn [rt] (some-> rt (update :cancel-fns (fnil conj []) f)))))
(defn release!
"Drops the runtime entry once the job settled. The persisted record stays
until its TTL."
[job-id]
(swap! registry dissoc (str job-id)))
(defn- live
"The job as the lifecycle last left it, or nil once it settled. Callers hold
the snapshot handed to them when their work started; writing that back would
resurrect a failed export as running and drop the error with it."
[job]
(when-let [current (:job (runtime (:id job)))]
(when-not (terminal? current)
current)))
(defn- persist-and-publish!
[job]
(store-job! job)
(publish! job)
(store/persist! job))
(defn transition!
"Moves the job on. The first terminal state wins: anything arriving after it
is dropped, so a late failure cannot overwrite a cancellation, nor a straggler
overwrite either."
[job data]
(if-let [job (live job)]
(persist-and-publish! (merge job data))
(p/resolved job)))
(defn start!
[job]
(transition! job {:state "running" :started-at (now-ms)}))
(defn progress!
"Reports `done` objects completed. Writes are throttled, so the caller need
not care how often it calls this."
[{:keys [id] :as job} done]
(if-let [job (live job)]
(let [k (str id)
now (now-ms)
last (:last-progress-ms (runtime id) 0)
job (assoc job :done done)
write? (>= (- now last) progress-throttle-ms)]
(store-job! job)
(if write?
(do
(swap! registry update k assoc :last-progress-ms now)
(persist-and-publish! job))
(p/resolved job)))
(p/resolved job)))
(defn complete!
[job {:keys [uri filename mtype size] :as _resource}]
(transition! job {:state "ended"
:ended-at (now-ms)
:done (:total job)
:resource-uri uri
:filename filename
:mtype mtype
:size size}))
(defn fail!
[job cause]
(l/error :hint "export job failed" :job-id (str (:id job)) :cause cause)
(transition! job {:state "error"
:ended-at (now-ms)
:error (ex-message cause)}))
(defn- cancel-local!
[job-id]
(let [k (str job-id)
rt (get @registry k)
job (:job rt)]
(if (or (nil? job) (terminal? job))
(p/resolved job)
(do
(swap! registry update k (fn [rt] (some-> rt (assoc :cancelled? true))))
(when-let [signal (:cancel-signal rt)]
(js/Atomics.store signal 0 1))
;; Recorded before the callbacks run, not after: one of them
;; (`scheduler/drop-queued!`) releases the job, and `transition!` on a
;; released job is a no-op, so a queued job would keep claiming to be
;; queued in the store and never publish its `cancelled` update.
(let [result (transition! job {:state "cancelled" :ended-at (now-ms)})]
(doseq [f (:cancel-fns rt)]
(try
(f)
(catch :default cause
(l/warn :hint "error on job cancel callback" :job-id k :cause cause))))
result)))))
(defn cancel!
"Cancels a job. One this process does not own is broadcast over the cancel
topic, so whoever runs it acts on it. Idempotent."
[job-id]
(if (some? (runtime job-id))
(cancel-local! job-id)
(->> (fetch job-id)
(p/mcat (fn [job]
(cond
(nil? job) (p/resolved nil)
(terminal? job) (p/resolved job)
:else (p/do
(store/request-cancel! (:id job))
job)))))))
(defn- clean-abandoned!
"Marks every job left mid-flight by a previous process as cancelled.
A queue and its running jobs live in the memory of the process that owns
them, so nothing in flight when it died can be resumed; without this the
record would keep claiming to be running until its TTL expires.
NOTE: the store cannot tell whose jobs are whose, so with more than one
exporter behind a load balancer this would also cancel a sibling's running
jobs. Single-instance deployments only."
[]
(->> (store/fetch-all)
(p/mcat (fn [jobs]
(let [abandoned (remove terminal? jobs)]
(when (seq abandoned)
(l/warn :hint "cancelling jobs abandoned by a previous process"
:count (count abandoned)))
(->> abandoned
(map (fn [job]
(store/persist! (assoc job
:state "cancelled"
:interrupted true
:ended-at (now-ms)))))
(p/all)))))
(p/fmap (fn [result] (count result)))
(p/merr (fn [cause]
(l/warn :hint "unable to clean abandoned jobs" :cause cause)
(p/resolved 0)))))
(defn init
[]
(store/on-cancel-request
(fn [job-id]
(when (some? (runtime job-id))
(l/info :hint "remote cancel request" :job-id job-id)
(cancel-local! job-id))))
(clean-abandoned!))

149
exporter/src/app/jobs/scheduler.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 INC Sucursal en España SL
(ns app.jobs.scheduler
"Admission control for export jobs.
Limits concurrent jobs and rejects work rather than allowing an unbounded backlog.
Queue order is FIFO, except jobs whose profile is already at its cap are skipped,
as are headless jobs once every render worker is busy."
(:require
[app.common.exceptions :as ex]
[app.common.logging :as l]
[app.config :as cf]
[app.jobs :as jobs]
[app.jobs.utils :as job.utils]
[app.wasm.pool :as pool]
[promesa.core :as p]))
(l/set-level! :debug)
(defonce ^:private state
(atom {:running {} ;; job-id -> {:profile-id :headless?}
:queue []})) ;; vector of {:job :resolve :reject}
(defn- max-concurrent [] (cf/get :exporter-max-concurrent-jobs 4))
(defn- max-per-profile [] (cf/get :exporter-max-jobs-per-profile 2))
(defn- max-queued [] (cf/get :exporter-queue-max 64))
(defn- headless?
[job]
(= "wasm" (:backend job)))
(defn- running-for
[{:keys [running]} profile-id]
(count (filter #(= profile-id (:profile-id %)) (vals running))))
(defn- running-headless
[{:keys [running]}]
(count (filter :headless? (vals running))))
(defn- eligible?
[state job]
(and (< (count (:running state)) (max-concurrent))
(< (running-for state (:profile-id job)) (max-per-profile))
;; A headless job holds one render worker for its whole run, so admitting
;; more of them than there are workers would only move the wait inside
;; the pool, with the job already reporting itself as running.
(or (not (headless? job))
(< (running-headless state) (pool/capacity)))))
(declare ^:private pump!)
(defn- finish!
[job-id]
(swap! state update :running dissoc (str job-id))
(jobs/release! job-id)
(job.utils/release! job-id)
(pump!))
(defn- execute!
[{:keys [id profile-id] :as job}]
(swap! state update :running assoc (str id) {:profile-id profile-id
:headless? (headless? job)})
(if (jobs/cancelled? id)
(do (finish! id)
(p/resolved job))
(let [run-fn (jobs/run-fn id)]
(->> (p/do (jobs/start! job))
(p/mcat (fn [job] (p/do (run-fn job))))
(p/fnly (fn [_ _] (finish! id)))))))
(defn- drop-queued!
"Removes a queued job and settles its promise, freeing its queue slot on cancellation."
[job-id]
(let [entry (volatile! nil)]
(swap! state (fn [state]
(let [queue (:queue state)
idx (->> (map-indexed vector queue)
(some (fn [[idx entry]]
(when (= (str job-id) (str (-> entry :job :id)))
idx))))]
(if idx
(do (vreset! entry (nth queue idx))
(assoc state :queue (into (subvec queue 0 idx) (subvec queue (inc idx)))))
state))))
(when-let [{:keys [resolve]} @entry]
(jobs/release! job-id)
(job.utils/release! job-id)
(resolve nil))))
(defn- take-eligible
"Pops the first queued entry that can run now, or nil."
[state]
(let [queue (:queue state)
idx (->> (map-indexed vector queue)
(some (fn [[idx entry]]
(when (eligible? state (:job entry))
idx))))]
(when idx
[(assoc state :queue (into (subvec queue 0 idx) (subvec queue (inc idx))))
(nth queue idx)])))
(defn- pump!
[]
(loop []
(let [entry (volatile! nil)]
(swap! state (fn [state]
(if-let [[next-state next-entry] (take-eligible state)]
(do (vreset! entry next-entry) next-state)
(do (vreset! entry nil) state))))
(when-let [{:keys [job resolve reject]} @entry]
(-> (execute! job)
(p/then resolve)
(p/catch reject))
(recur)))))
(defn submit!
"Registers `job` for execution. Returns a promise of the job's result, which
resolves when the export actually finishes; callers that only need the handle
can ignore it. Raises when the exporter is saturated."
[job]
(let [resolve* (volatile! nil)
reject* (volatile! nil)
pending (p/create (fn [resolve reject]
(vreset! resolve* resolve)
(vreset! reject* reject)))]
(if (eligible? @state job)
(-> (execute! job)
(p/then @resolve*)
(p/catch @reject*))
(let [queued (volatile! false)]
(swap! state (fn [state]
(if (< (count (:queue state)) (max-queued))
(do (vreset! queued true)
(update state :queue conj {:job job
:resolve @resolve*
:reject @reject*}))
(do (vreset! queued false) state))))
(when-not @queued
(ex/raise :type :validation
:code :queue-full
:hint "too many queued export jobs"))
(jobs/on-cancel (:id job) (fn [] (drop-queued! (:id job))))
(l/dbg :hint "export job queued" :job-id (str (:id job)))))
pending))

87
exporter/src/app/jobs/store.cljs Normal file
View File

@ -0,0 +1,87 @@
;; This Source Code Form is subject to the terms of the Mozilla Public
;; License, v. 2.0. If a copy of the MPL was not distributed with this
;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
;;
;; Copyright (c) KALEIDOS INC Sucursal en España SL
(ns app.jobs.store
"Redis persistence for export jobs.
Stores each job as a single blob with a TTL matching the exported file,
so records expire with their files.
Reads use Redis. Cancellation requires the process running the job."
(:require
[app.common.logging :as l]
[app.common.transit :as t]
[app.config :as cf]
[app.redis :as redis]
[promesa.core :as p]))
(def ^:private cancel-topic (redis/->key "job-cancel"))
(defn- job-key
[job-id]
(redis/->key "job." job-id))
(defn- ttl
[]
(cf/get :exporter-job-ttl 3600))
(defn persist!
"Writes the job record and refreshes its TTL.
If the write fails, log it and continue. The export still runs
and publishes websocket updates, but the job can't be fetched
afterward (fetch returns nil, REST returns 404)."
[{:keys [id state] :as job}]
(let [jkey (job-key id)]
(->> (p/do
(redis/hset! jkey {:data (t/encode-str job)})
(redis/expire! jkey (ttl))
job)
(p/merr (fn [cause]
(if (= :redis-not-available (:code (ex-data cause)))
(l/warn :hint "job record not persisted, no redis connection"
:job-id (str id) :state state)
(l/error :hint "unable to persist job record"
:job-id (str id) :state state :cause cause))
(p/resolved job))))))
(defn fetch
"The job record, or nil when unknown or expired."
[job-id]
(->> (redis/hgetall (job-key job-id))
(p/fmap (fn [data]
(when-let [blob (get data "data")]
(try
(t/decode-str blob)
(catch :default cause
(l/warn :hint "unable to decode job record" :job-id (str job-id) :cause cause)
nil)))))))
(defn fetch-all
[]
(->> (redis/scan (redis/->key "job.*"))
(p/mcat (fn [keys]
(->> (map (fn [k]
(->> (redis/hgetall k)
(p/fmap (fn [data]
(when-let [blob (get data "data")]
(try
(t/decode-str blob)
(catch :default _ nil)))))))
keys)
(p/all))))
(p/fmap (fn [jobs] (vec (remove nil? jobs))))))
(defn request-cancel!
"Asks every exporter to cancel `job-id`. Only the one running it will act."
[job-id]
(redis/pub! cancel-topic (str job-id)))
(defn on-cancel-request
"Registers `handler` (fn of the job-id string) for cancel requests. Returns an
unsubscribe fn."
[handler]
(redis/sub! cancel-topic handler))

86
exporter/src/app/jobs/utils.cljs Normal file
View File

@ -0,0 +1,86 @@
;; This Source Code Form is subject to the terms of the Mozilla Public
;; License, v. 2.0. If a copy of the MPL was not distributed with this
;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
;;
;; Copyright (c) KALEIDOS INC Sucursal en España SL
(ns app.jobs.utils
"Temp file ownership for export jobs.
Temp files used to be cleaned only by the per-file timer in `app.util.shell`,
an hour after creation and lost entirely on restart. Here each job owns the
paths it creates, so they are dropped as soon as it settles and whatever a
crash left behind is cleaned at boot."
(:require
["node:fs/promises" :as fsp]
["node:path" :as path]
[app.common.logging :as l]
[app.config :as cf]
[app.util.shell :as sh]
[cuerdas.core :as str]
[promesa.core :as p]))
(def ^:private managed-prefix "penpot.")
(defonce ^:private tracked (atom {}))
(defn track!
"Registers `path` as owned by `job-id`, so it is removed when the job settles."
[job-id path]
(when (and job-id path)
(swap! tracked update (str job-id) (fnil conj #{}) path))
path)
(defn- remove-path!
[path]
(->> (p/do (fsp/rm path #js {:recursive true :force true}))
(p/merr (fn [cause]
(l/warn :hint "unable to remove job temp file" :path path :cause cause)
(p/resolved nil)))))
(defn release!
"Removes every file the job owns. Called once the job reached a terminal
state and its result has already been uploaded, so nothing else reads them."
[job-id]
(let [k (str job-id)
paths (get @tracked k)]
(swap! tracked dissoc k)
(if (seq paths)
(->> (map remove-path! paths)
(p/all)
(p/fmap (fn [_]
(l/dbg :hint "released job temp files" :job-id k :count (count paths))
nil)))
(p/resolved nil))))
(defn- clean!
"Removes managed temp files older than the job TTL. They can only be leftovers
of a previous process: every live one belongs to a job of this process."
[]
(let [max-age (* 1000 (cf/get :exporter-job-ttl 3600))
now (js/Date.now)]
(->> (p/do (fsp/readdir sh/tmpdir))
(p/mcat (fn [entries]
(->> (filter #(str/starts-with? % managed-prefix) entries)
(map (fn [entry]
(let [fpath (path/join sh/tmpdir entry)]
(->> (p/do (fsp/stat fpath))
(p/mcat (fn [^js stat]
(if (> (- now (inst-ms (.-mtime stat))) max-age)
(->> (remove-path! fpath)
(p/fmap (constantly 1)))
(p/resolved 0))))
(p/merr (fn [_] (p/resolved 0)))))))
(p/all))))
(p/fmap (fn [results]
(let [removed (reduce + 0 results)]
(when (pos? removed)
(l/info :hint "removed orphaned export temp files" :count removed))
removed)))
(p/merr (fn [cause]
(l/warn :hint "temp file cleanup failed" :cause cause)
(p/resolved 0))))))
(defn init
[]
(clean!))

149
exporter/src/app/redis.cljs
View File

@ -8,47 +8,170 @@
(:require
["ioredis" :as redis]
[app.common.data.macros :as dm]
[app.common.exceptions :as ex]
[app.common.logging :as l]
[app.common.transit :as t]
[app.config :as cf]))
[app.config :as cf]
[promesa.core :as p]))
(l/set-level! :trace)
(def client (atom nil))
;; A connection in subscriber mode rejects every other command, so the
;; subscriptions need a connection of their own.
(def ^:private subscriber (atom nil))
(def ^:private subscriptions (atom {}))
(defn- create-client
[uri]
[uri role]
(let [^js client (new redis/default uri)]
(.on client "connect"
(fn [] (l/info :hint "redis connection established" :uri uri)))
(fn [] (l/info :hint "redis connection established" :uri uri :role role)))
(.on client "error"
(fn [cause] (l/error :hint "error on redis connection" :cause cause)))
(fn [cause] (l/error :hint "error on redis connection" :role role :cause cause)))
(.on client "close"
(fn [] (l/warn :hint "connection closed")))
(fn [] (l/warn :hint "connection closed" :role role)))
(.on client "reconnect"
(fn [ms] (l/warn :hint "reconnecting to redis" :ms ms)))
(fn [ms] (l/warn :hint "reconnecting to redis" :role role :ms ms)))
(.on client "end"
(fn [] (l/warn :hint "client ended, no more connections will be attempted")))
(fn [] (l/warn :hint "client ended, no more connections will be attempted" :role role)))
client))
(defn- dispatch-message
[topic payload]
(doseq [handler (get @subscriptions topic)]
(try
(handler payload)
(catch :default cause
(l/error :hint "error on redis subscription handler" :topic topic :cause cause)))))
(defn init
[]
(swap! client (fn [prev]
(when prev (.disconnect ^js prev))
(create-client (cf/get :redis-uri)))))
(let [uri (cf/get :redis-uri)]
(swap! client (fn [prev]
(when prev (.disconnect ^js prev))
(create-client uri "commands")))
(swap! subscriber (fn [prev]
(when prev (.disconnect ^js prev))
(let [^js conn (create-client uri "subscriber")]
(.on conn "message" (fn [topic payload] (dispatch-message topic payload)))
;; Reinstate subscriptions after a reconnection.
(.on conn "connect"
(fn []
(doseq [topic (keys @subscriptions)]
(.subscribe conn topic))))
conn)))))
(defn stop
[]
(reset! subscriptions {})
(swap! subscriber (fn [conn]
(when conn (.quit ^js conn))
nil))
(swap! client (fn [client]
(when client (.quit ^js client))
nil)))
(def ^:private tenant (cf/get :tenant))
(defn ->tenant-key
"Namespaces `parts` under the tenant, the prefix the backend msgbus uses."
[& parts]
(dm/str tenant "." (apply str parts)))
(defn ->key
"Namespaces `parts` under the exporter, inside the tenant."
[& parts]
(dm/str "penpot.exporter." tenant "." (apply str parts)))
(defn pub!
"Publishes on `topic`, which must already be namespaced."
[topic payload]
(let [payload (if (map? payload) (t/encode-str payload) payload)
topic (dm/str tenant "." topic)]
(let [payload (if (map? payload) (t/encode-str payload) payload)]
(when-let [client @client]
(.publish ^js client topic payload))))
(defn sub!
"Subscribes `handler` (fn of the raw payload string) to `topic`, which must
already be namespaced. Returns a 0-arg fn that removes this handler."
[topic handler]
(swap! subscriptions update topic (fnil conj []) handler)
(when-let [conn @subscriber]
(.subscribe ^js conn topic))
(fn []
(swap! subscriptions update topic (fn [handlers] (vec (remove #(= % handler) handlers))))))
(defn- with-client
"Runs `f` against the command connection. Rejects when there is no connection
or the command fails: whether a failure is survivable depends on what the
caller was doing, and only the caller knows."
[f]
(if-let [client @client]
(p/do (f client))
(p/rejected (ex/error :type :internal
:code :redis-not-available
:hint "no redis connection"))))
(defn- with-client-lenient
"For reads, where an unreachable redis is reported as \"nothing there\"."
[f]
(->> (with-client f)
(p/merr (fn [cause]
(l/warn :hint "redis command failed" :cause cause)
(p/resolved nil)))))
(defn hset!
"Writes `data` (a map of string/keyword -> value) as a hash. Nil values are
dropped, since redis has no null."
[k data]
(let [obj (reduce-kv (fn [obj field value]
(if (some? value)
(doto obj (unchecked-set (name field) (str value)))
obj))
#js {}
data)]
(if (zero? (alength (js/Object.keys obj)))
(p/resolved nil)
(with-client (fn [^js client] (.hset client k obj))))))
(defn hgetall
"Returns the hash as a map of string keys, or nil when it does not exist."
[k]
(->> (with-client-lenient (fn [^js client] (.hgetall client k)))
(p/fmap (fn [result]
(when (and result (pos? (alength (js/Object.keys result))))
(persistent!
(reduce (fn [res field]
(assoc! res field (unchecked-get result field)))
(transient {})
(js/Object.keys result))))))))
(defn expire!
[k seconds]
(with-client (fn [^js client] (.expire client k seconds))))
(defn del!
[k]
(with-client (fn [^js client] (.del client k))))
(defn scan
"Every key matching `pattern`, walked in cursor batches so a large keyspace is
never blocked the way `KEYS` would block it.
Batches are accumulated in memory rather than consumed as a stream, which a
promise-returning fn cannot express. Fine for the job keyspace, but reading
redis wants a streaming or reactive interface before it is used for more."
[pattern]
(letfn [(step [cursor found]
(->> (with-client-lenient (fn [^js client] (.scan client cursor "MATCH" pattern "COUNT" 200)))
(p/mcat (fn [result]
(if (nil? result)
(p/resolved found)
(let [next-cursor (aget result 0)
found (into found (aget result 1))]
(if (= "0" next-cursor)
(p/resolved found)
(step next-cursor found))))))))]
(step "0" [])))

29
exporter/src/app/renderer.cljs
View File

@ -26,6 +26,7 @@
(s/def ::token ::us/string)
(s/def ::filename ::us/string)
(s/def ::is-wasm ::us/boolean)
(s/def ::job-id ::us/uuid)
(s/def ::object
(s/keys :req-un [::id ::name ::suffix ::filename]
@ -36,18 +37,22 @@
(s/def ::render-params
(s/keys :req-un [::file-id ::page-id ::scale ::token ::type ::objects]
:opt-un [::is-wasm]))
:opt-un [::is-wasm ::job-id]))
(defn headless?
"Whether `params` renders with render-wasm rather than a browser."
[{:keys [type is-wasm]}]
(and is-wasm (contains? cf/flags :wasm-export) (not= :svg type)))
(defn render
[{:keys [type is-wasm] :as params} on-object]
(us/verify ::render-params params)
(us/verify fn? on-object)
(let [wasm-export? (contains? cf/flags :wasm-export)
headless? (and is-wasm wasm-export? (not= :svg type))]
(let [headless? (headless? params)]
(when is-wasm
(l/info :hint "render"
:type type
:wasm-export wasm-export?
:wasm-export (contains? cf/flags :wasm-export)
:backend (if headless? "wasm" "browser")))
(if headless?
(rw/render params on-object)
@ -58,3 +63,19 @@
:pdf (rp/render params on-object)
:svg (rs/render params on-object)))))
(defn with-scope
"Runs `f`, a fn of a render fn with the same signature as `render`. Exports
that render headless share one worker for the whole call instead of acquiring
one per render; the browser backend keeps rendering them in parallel."
[exports f]
(if (some headless? exports)
(rw/with-scope (:job-id (first exports))
(fn [render-leased]
(f (fn [params on-object]
(us/verify ::render-params params)
(us/verify fn? on-object)
(if (headless? params)
(render-leased params on-object)
(render params on-object))))))
(f render)))

469
exporter/src/app/renderer/wasm.cljs
View File

@ -5,447 +5,46 @@
;; Copyright (c) KALEIDOS SUBSIDIARY SL
(ns app.renderer.wasm
"Headless renderer backend: renders exports with the render-wasm Skia
pipeline in this Node process, with no browser and no WebGL.
"Main-thread side of the headless renderer.
Per request: fetch scene (get-page RPC) -> serialize -> provision fonts and
images -> relayout text with the real fonts -> render each object.
One shared WASM design state, so requests are serialized one at a time.
Handles png/jpeg/webp (Skia encodes all three) and pdf; `:svg` stays on the
browser path."
Renders run on pooled workers because Skia calls are synchronous and would block
the HTTP server and other exports. Each job keeps one worker for all its renders,
sharing its caches and pool slot."
(:require
["node:fs" :as fs]
["undici" :as http]
[app.common.data :as d]
[app.common.fonts :as cfnt]
;; Required for side effects: these register the transit read handlers and
;; deftype impls the `get-page` response is decoded into.
[app.common.geom.matrix]
[app.common.geom.point]
[app.common.geom.rect]
[app.common.logging :as l]
[app.common.transit :as t]
[app.common.types.fills.impl]
[app.common.types.objects-map]
[app.common.types.path.impl]
[app.common.types.shape]
[app.common.types.shape.images :as images]
[app.common.uri :as u]
[app.common.uuid :as uuid]
[app.config :as cf]
[app.util.mime :as mime]
[app.util.shell :as sh]
[app.wasm :as wasm]
[app.wasm.serialize :as serialize]
[cuerdas.core :as str]
[app.jobs :as jobs]
[app.wasm.pool :as pool]
[promesa.core :as p]))
;; --- module lifecycle (one shared, lazily-initialized instance)
(defonce ^:private module* (atom nil))
(defn- ensure-module!
(defn- serializer
"Chains thunks so a job's renders run one at a time on its worker. A failure
is isolated: it doesn't break the chain for the next one."
[]
(or @module*
(reset! module* (wasm/init!))))
(let [queue (atom (p/resolved nil))]
(fn [thunk]
(let [result (p/handle @queue (fn [_ _] (thunk)))]
(reset! queue (p/handle result (fn [_ _] nil)))
result))))
;; --- serialized access to the shared module
;;
;; `handle-multiple-export` fans out partitions concurrently, but there is one
;; design state and one global mem buffer, so their serialize/render/alloc must
;; not interleave.
(defonce ^:private queue (atom (p/resolved nil)))
(defn- enqueue!
"Runs `thunk` (0-arg, returns a promise) only after all previously enqueued
work has settled. Returns `thunk`'s promise. A task's failure is isolated:
it doesn't break the chain for the next task."
[thunk]
(let [result (p/handle @queue (fn [_ _] (thunk)))]
(reset! queue (p/handle result (fn [_ _] nil)))
result))
;; --- backend endpoints
;;
;; Every fetch targets the internal endpoint (falling back to public-uri),
;; in a deployment the exporter reaches the backend over the container network
(defn- internal-uri
"Absolute URI for `path` on the internal (backend) endpoint."
[path]
(-> (cf/get-internal-uri)
(u/ensure-path-slash)
(u/join path)
(str)))
(defn- error-detail
"Node's fetch reports every transport failure as a bare `TypeError: fetch
failed`; the actual reason (TLS rejection, DNS, ECONNREFUSED) is buried in a
nested `cause` chain that the logger does not print. Flattens the chain into
one readable string."
[cause]
(->> (iterate (fn [^js e] (unchecked-get e "cause")) cause)
(take-while some?)
(take 5)
(map (fn [^js e]
(let [code (unchecked-get e "code")
msg (or (unchecked-get e "message") (str e))]
(if code (str code ": " msg) msg))))
(str/join " <- ")))
(defn- fetch!
"`undici/fetch` that fails with an ex-info carrying the target uri and the
unwrapped cause chain, so a failed request says what actually went wrong and
against which endpoint."
[uri opts]
(->> (p/do (http/fetch uri opts))
(p/merr (fn [cause]
(p/rejected (ex-info "http fetch failed"
{:uri uri :detail (error-detail cause)}
cause))))))
(defn- explain
"Log-friendly reason for `cause`: the detail `fetch!` already attached, or a
freshly unwrapped chain for anything else (WASM aborts, decode errors)."
[cause]
(or (:detail (ex-data cause))
(error-detail cause)))
(defn- rpc-headers
"Auth headers for backend RPC calls (management key + bearer)."
[token]
#js {"Content-Type" "application/transit+json"
"X-Shared-Key" (str "exporter " cf/management-key)
"Authorization" (str "Bearer " token)})
(defn- asset-headers
"Auth headers for `/assets/*`. Cookie, not Bearer: those endpoints redirect to
a presigned S3/minio URL, and a Bearer header makes S3 400 (\"multiple
authentication types\")."
[token]
#js {"X-Shared-Key" (str "exporter " cf/management-key)
"Cookie" (str "auth-token=" token)})
;; --- shape bundle fetch (backend RPC)
(defn- fetch-objects
"Fetches the exported roots and their children from the backend via the
`get-page` RPC (`:object-id`, as the browser render path does), using the
same auth the exporter uses elsewhere (management key + bearer)."
[{:keys [file-id page-id share-id token objects]}]
(let [headers (rpc-headers token)
root-ids (into #{} (map :id) objects)
body (t/encode-str (cond-> {:file-id file-id
:page-id page-id}
(seq root-ids) (assoc :object-id root-ids)
share-id (assoc :share-id share-id)))
uri (internal-uri "api/rpc/command/get-page")]
(l/dbg :hint "wasm render: get-page"
:uri uri
:file-id (str file-id)
:page-id (str page-id)
:roots (count root-ids))
(->> (fetch! uri #js {:method "POST" :headers headers :body body})
(p/mcat (fn [^js resp]
(if (= 200 (.-status resp))
(.text resp)
(->> (.text resp)
(p/mcat (fn [resp-body]
(l/error :hint "wasm render: get-page failed"
:uri uri
:status (.-status resp)
:body resp-body)
(p/rejected (ex-info "get-page failed"
{:status (.-status resp)
:body resp-body}))))))))
(p/fmap t/decode-str)
(p/fmap :objects))))
;; --- font resolution
;;
;; The text serializer keeps each font's real uuid, so `wasm/fonts-for-shape`
;; reports it. Custom (team) fonts resolve through the file's font variants,
;; google fonts through the shared `app.common.fonts` catalog; builtin
;; fonts through its bundled family + the frontend's static `/fonts/`.
(defn- fetch-font-variants
"Team (custom) font variants for the file, or nil — a failure here degrades
to fallback fonts, it does not fail the export."
[{:keys [file-id share-id token]}]
(let [headers (rpc-headers token)
body (t/encode-str (cond-> {:file-id file-id}
share-id (assoc :share-id share-id)))
uri (internal-uri "api/rpc/command/get-font-variants")]
(->> (fetch! uri #js {:method "POST" :headers headers :body body})
(p/mcat (fn [^js resp]
(if (= 200 (.-status resp))
(.text resp)
(p/resolved nil))))
(p/fmap (fn [s] (when s (t/decode-str s))))
(p/merr (fn [cause]
(l/warn :hint "wasm render: get-font-variants failed"
:uri uri :detail (explain cause) :cause cause)
(p/resolved nil))))))
(defn- fetch-ttf-bytes
"Downloads a TTF, returning a promise of an ArrayBuffer (or nil). A failure
here degrades to fallback fonts, it does not fail the export."
([uri] (fetch-ttf-bytes uri #js {:method "GET"}))
([uri opts]
(->> (fetch! uri opts)
(p/mcat (fn [^js resp]
(if (= 200 (.-status resp))
(.arrayBuffer resp)
(p/resolved nil))))
(p/merr (fn [cause]
(l/warn :hint "wasm render: font fetch failed"
:uri uri :detail (explain cause) :cause cause)
(p/resolved nil))))))
;; TTF bytes cached for the process lifetime, keyed by whatever identifies the
;; variant (a gfont id+weight+style, a builtin file name).
(defonce ^:private font-bytes* (atom {}))
(defn- cached-ttf-bytes
[cache-key fetch-fn]
(if-let [bytes (get @font-bytes* cache-key)]
(p/resolved bytes)
(->> (fetch-fn)
(p/fmap (fn [buf]
(when buf (swap! font-bytes* assoc cache-key buf))
buf)))))
(defn- fetch-asset-bytes
[asset-id {:keys [token]}]
(fetch-ttf-bytes (internal-uri (str "assets/by-id/" asset-id))
#js {:method "GET" :headers (asset-headers token)}))
(defn- fetch-gfont-bytes
[ttf-url]
(fetch-ttf-bytes (cfnt/gstatic->proxy-url ttf-url (internal-uri "internal/gfonts/font"))))
(defn- fetch-builtin-font-bytes
[ttf-file]
(cached-ttf-bytes ttf-file #(fetch-ttf-bytes (internal-uri (str "fonts/" ttf-file)))))
(defn- make-resolve-font
"Builds a `resolve-font` fn (family map -> promise of TTF bytes). Custom
variants first, matching uuid+weight+style then degrading to uuid+weight then
uuid; the bundled fonts for `uuid/zero`, which is what `font-id->uuid` maps
every builtin family to; google catalog otherwise."
[variants params]
(fn [{:keys [id weight style]}]
(let [font-uuid (uuid/from-unsigned-parts (aget id 0) (aget id 1) (aget id 2) (aget id 3))
style-str (if (zero? style) "normal" "italic")
variant (or (d/seek (fn [v] (and (= (:font-id v) font-uuid)
(= (:font-weight v) weight)
(= (name (:font-style v)) style-str)))
variants)
(d/seek (fn [v] (and (= (:font-id v) font-uuid)
(= (:font-weight v) weight)))
variants)
(d/seek (fn [v] (= (:font-id v) font-uuid)) variants))]
(cond
(:ttf-file-id variant)
(fetch-asset-bytes (:ttf-file-id variant) params)
(= uuid/zero font-uuid)
(fetch-builtin-font-bytes (cfnt/resolve-ttf-file weight style))
:else
(if-let [gurl (cfnt/resolve-ttf-url font-uuid weight style)]
(fetch-gfont-bytes gurl)
(p/resolved nil))))))
;; --- fallback fonts (emoji + per-script noto fonts)
;;
;; Emoji and non-latin scripts render through fallback families, not through
;; any span's font family, so `wasm/fonts-for-shape` never reports them and the
;; provisioning above never uploads them. Must run per request, since
;; `clear-fonts!` empties the store; the TTF bytes stay cached per process.
(defn- scene-fallback-fonts
"Fallback font descriptors needed by the scene's text. Deduped because
several languages map to one noto family and provisioning is concurrent —
otherwise they all miss the byte cache at once and refetch the same TTF."
[scene]
(let [texts (for [shape (vals scene)
:when (= :text (:type shape))
node (or (some->> (:content shape) (tree-seq :children :children)) [])
:let [text (:text node)]
:when (string? text)]
text)
emoji? (boolean (some cfnt/contains-emoji? texts))
langs (reduce cfnt/collect-used-languages #{} texts)]
(distinct
(cond-> (cfnt/add-noto-fonts [] langs)
emoji? (cfnt/add-emoji-font)))))
(defn- fetch-fallback-font-bytes
"Downloads one fallback font's TTF. Cached by the whole variant, not just
`font-id`: `resolve-ttf-url` picks a different TTF per weight/style, so a
font-id-only key would serve the first downloaded variant for every other one."
[{:keys [font-id weight style]}]
(if-let [ttf-url (some-> (cfnt/gfont-id->uuid font-id) (cfnt/resolve-ttf-url weight style))]
(cached-ttf-bytes [font-id weight style] #(fetch-gfont-bytes ttf-url))
(p/resolved nil)))
(defn- provision-fallback-fonts!
[scene]
(->> (scene-fallback-fonts scene)
(map (fn [{:keys [font-id weight style is-emoji is-fallback] :as font}]
(if-let [font-uuid (cfnt/gfont-id->uuid font-id)]
(->> (fetch-fallback-font-bytes font)
(p/fmap (fn [buf]
(if buf
(wasm/store-font! {:id (uuid/get-u32 font-uuid)
:weight weight
:style style
:emoji? (boolean is-emoji)
:fallback? (boolean is-fallback)}
buf)
(l/warn :hint "wasm render: fallback font unavailable"
:font-id font-id)))))
(p/resolved nil))))
(p/all)))
;; --- image resolution
;;
;; Image fills reference file-media ids; the encoded bytes go straight to
;; `_store_image` (Skia decodes, no WebGL), keyed by media uuid so this happens
;; once per request rather than per rendered object.
(defn- fetch-file-media-bytes
"Downloads an image fill's encoded bytes by file-media id."
[media-id {:keys [token]}]
(let [headers (asset-headers token)
uri (internal-uri (str "assets/by-file-media-id/" media-id))]
(->> (fetch! uri #js {:method "GET" :headers headers})
(p/mcat (fn [^js resp]
(if (= 200 (.-status resp))
(.arrayBuffer resp)
(do
(l/warn :hint "wasm render: image fetch non-200"
:media-id (str media-id)
:uri uri
:status (.-status resp))
(p/resolved nil)))))
(p/merr (fn [cause]
(l/warn :hint "wasm render: image fetch failed"
:media-id (str media-id) :uri uri
:detail (explain cause) :cause cause)
(p/resolved nil))))))
(defn- provision-images!
"Fetches and stores every image the scene references (shape, stroke and
text-span fills, enumerated by `app.common.types.shape.images`). Unlike fonts,
the image store is not reset per request, so already-held images are skipped
and repeated exports of a file reuse them."
[scene params]
(let [all-ids (images/scene-image-ids scene)
new-ids (remove wasm/image-cached? all-ids)]
(l/dbg :hint "wasm render: provisioning images"
:total (count all-ids)
:cached (- (count all-ids) (count new-ids)))
(->> new-ids
(map (fn [image-id]
(->> (fetch-file-media-bytes image-id params)
(p/fmap (fn [buf]
(if buf
(do
(l/dbg :hint "wasm render: image stored"
:media-id (str image-id)
:bytes (.-byteLength ^js buf))
(wasm/store-image! image-id buf))
(l/warn :hint "wasm render: image unavailable"
:media-id (str image-id))))))))
(p/all))))
(defn- relayout-text!
"Recomputes layout for every text shape, once the real fonts are provisioned
(serialize-time layout used the fallback)."
[scene]
(doseq [shape (vals scene)
:when (= :text (:type shape))]
(wasm/update-text-layout! (:id shape))))
;; --- render
(defn- render-object-bytes
[type id scale]
(if (= :pdf type)
(let [bytes (wasm/render-shape-pdf id scale)]
(l/dbg :hint "PDF generated via Skia (render-wasm headless)"
:object-id (str id)
:backend "skia-wasm"
:bytes (.-length bytes))
bytes)
(wasm/render-shape-raster id scale type)))
(defn- render*
[{:keys [scale type objects] :as params} on-object]
(l/dbg :hint "wasm render: start"
:type type
:scale scale
:objects (count objects)
:file-id (str (:file-id params))
:page-id (str (:page-id params)))
(->> (ensure-module!)
(p/mcat (fn [_] (fetch-objects params)))
(p/mcat (fn [scene]
(l/dbg :hint "wasm render: scene fetched" :shapes (count scene))
(serialize/serialize-scene! scene)
(l/dbg :hint "wasm render: scene serialized")
;; So fonts from a previous request don't leak into this one.
(wasm/clear-fonts!)
(->> (p/all [(fetch-font-variants params)
(provision-images! scene params)
(provision-fallback-fonts! scene)])
(p/mcat
(fn [[variants _]]
(let [resolve-font (make-resolve-font (or variants []) params)]
;; Before rendering, so the relayout below sees real
;; font metrics. Deduped across objects: a partition
;; sharing one family downloads its TTF once.
(wasm/provision-fonts! (map :id objects) resolve-font))))
(p/mcat
(fn [_]
(relayout-text! scene)
(p/run
(fn [{:keys [id] :as object}]
(let [bytes (render-object-bytes type id scale)
path (sh/tempfile :prefix "penpot.tmp.wasm."
:suffix (mime/get-extension type))]
(l/dbg :hint "wasm render: object rendered"
:object-id (str id) :bytes (.-length bytes))
(fs/writeFileSync path bytes)
;; `on-object` returns a plain value (zip append) or
;; a promise (single export's file move); `p/do`
;; normalizes both to a thenable.
(p/do (on-object (assoc object :path path)))))
objects))))))
(p/fmap (fn [result]
;; After the request, never mid-render, so an image can't
;; disappear under a running export.
(let [evicted (wasm/evict-images! wasm/image-cache-mb)]
(when (pos? evicted)
(l/info :hint "wasm render: evicted cached images" :count evicted)))
result))
(p/merr (fn [cause]
(l/error :hint "wasm render: failed"
:detail (explain cause)
:internal-uri (str (cf/get-internal-uri))
:cause cause)
;; A panic can leave the mem buffer allocated or the instance
;; aborted; drop it so the next request rebuilds a fresh one.
(reset! module* nil)
(p/rejected cause)))))
(defn with-scope
"Runs `f`, a fn of a 2-arg render fn. Every render goes to the same worker,
one at a time, so the cancel check runs as each render's turn comes up."
[job-id f]
(pool/with-worker
(fn [worker]
(let [chain (serializer)
live (volatile! worker)
signal (when job-id (jobs/cancel-signal job-id))
opts {:cancel-buffer (some-> signal (.-buffer))
:cancelled? (when job-id #(jobs/cancelled? job-id))}]
(when job-id
;; Between objects the worker sees the flag; inside a render only
;; terminating the thread stops it. Cleared on the way out so a later
;; cancel cannot terminate a worker that is by then somebody else's.
(jobs/on-cancel job-id (fn [] (pool/terminate! @live))))
(->> (p/do (f (fn [params on-object]
(chain #(pool/render-on worker params on-object opts)))))
(p/fnly (fn [_ _] (vreset! live nil))))))))
(defn render
"Public entry. `enqueue!` keeps concurrent exports off each other's toes on
the shared WASM instance."
[params on-object]
(enqueue! (fn [] (render* params on-object))))
(with-scope (:job-id params) (fn [render*] (render* params on-object))))

61
exporter/src/app/router.cljs Normal file
View File

@ -0,0 +1,61 @@
;; This Source Code Form is subject to the terms of the Mozilla Public
;; License, v. 2.0. If a copy of the MPL was not distributed with this
;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
;;
;; Copyright (c) KALEIDOS INC Sucursal en España SL
(ns app.router
"Method + path dispatch.
Requests arrive with whatever prefix the proxy in front uses (`/api/export`
in devenv, `/` when talking to the process directly), so routes are matched
on the remainder after that prefix."
(:require
[app.common.exceptions :as ex]
[app.handlers.jobs :as jobs.handlers]
[cuerdas.core :as str]))
(def ^:private mount-point "/api/export")
(defn- route-path
[path]
(let [path (or path "/")
path (if (str/starts-with? path mount-point)
(subs path (count mount-point))
path)
path (str/rtrim path "/")]
(if (str/empty? path) "/" path)))
(defn- job-id
[path prefix]
(let [id (subs path (count prefix))]
(when-not (or (str/empty? id) (str/includes? id "/"))
id)))
(defn create
"Builds the request handler. `legacy-handler` serves the original
`POST /api/export` command multiplex."
[legacy-handler]
(fn [{:keys [:request/method :request/path] :as exchange}]
(let [path (route-path path)]
(cond
(and (= "post" method) (= "/" path))
(legacy-handler exchange)
(and (= "post" method) (= "/jobs" path))
(jobs.handlers/create exchange)
(and (= "get" method) (str/starts-with? path "/jobs/"))
(if-let [id (job-id path "/jobs/")]
(jobs.handlers/fetch exchange id)
(ex/raise :type :not-found :code :object-not-found :hint "unknown route"))
(and (= "delete" method) (str/starts-with? path "/jobs/"))
(if-let [id (job-id path "/jobs/")]
(jobs.handlers/cancel exchange id)
(ex/raise :type :not-found :code :object-not-found :hint "unknown route"))
:else
(ex/raise :type :not-found
:code :route-not-found
:hint (str "no route for " method " " path))))))

28
exporter/src/app/wasm.cljs
View File

@ -43,9 +43,9 @@
path in devenv and inside the bundle, so it is a constant."
"resources/wasm")
(def image-cache-mb
"Byte budget (MB) the image store is trimmed to between requests."
256)
(def image-cache-size
"Byte budget the image store is trimmed to between requests."
(* 256 1024 1024))
(defn- read-result-bytes
"Reads `len` bytes from the WASM heap starting at `offset`, copying them out
@ -121,9 +121,9 @@
[(aget id 0) (aget id 1) (aget id 2) (aget id 3) weight style])
(defn fonts-for-shapes
"Distinct font families needed by every subtree in `shape-ids`. Objects in a
partition overwhelmingly share families, so deduping here means one download
and one `_store_font` per family rather than one per object."
"Distinct font families needed by every subtree in `shape-ids`. Objects in one
export overwhelmingly share families, so deduping here means one download and
one `_store_font` per family rather than one per object."
[shape-ids]
(into [] (comp (mapcat fonts-for-shape)
(d/distinct-xf font-key))
@ -156,10 +156,14 @@
"Recomputes a text shape's layout with the currently provisioned fonts. Text is
laid out at serialize time using the fallback font (real fonts aren't uploaded
yet), so this must run again after `provision-fonts!` or glyph metrics/line
breaks are wrong."
breaks are wrong.
Forced, because provisioning a font changes nothing `update_layout` keys on:
it early-returns while the content is unchanged and the layout still matches
its container, which is exactly the case here."
[shape-id]
(let [buf (uuid/get-u32 shape-id)]
(h/call wasm/internal-module "_update_shape_text_layout_for"
(h/call wasm/internal-module "_force_update_shape_text_layout_for"
(aget buf 0) (aget buf 1) (aget buf 2) (aget buf 3))))
(defn image-cached?
@ -205,10 +209,10 @@
(h/call module "_store_image")))
(defn evict-images!
"Evicts least-recently-used images until the store retains at most `max-mb`
megabytes. Returns the number evicted."
[max-mb]
(h/call wasm/internal-module "_evict_images_to_budget" max-mb))
"Evicts least-recently-used images until the store retains at most `max-bytes`
bytes. Returns the number evicted."
[max-bytes]
(h/call wasm/internal-module "_evict_images_to_budget" max-bytes))
(defn provision-fonts!
"Resolves and uploads every font needed by `shape-ids`, each family fetched

246
exporter/src/app/wasm/pool.cljs Normal file
View File

@ -0,0 +1,246 @@
;; This Source Code Form is subject to the terms of the Mozilla Public
;; License, v. 2.0. If a copy of the MPL was not distributed with this
;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
;;
;; Copyright (c) KALEIDOS INC Sucursal en España SL
(ns app.wasm.pool
"Pool of headless render workers.
Mirrors `app.browser`: a `generic-pool` whose objects are `worker_threads`
instead of browsers, so acquisition and eviction behave the same way for both
render backends. A worker is expensive to build (it boots its own render-wasm
module), hence the pooling.
Acquisition is not capped: the admission scheduler is the backpressure, and
the idle watchdog guarantees a wedged worker gives its slot back.
Workers run the same bundle as the main thread; `app.core/start` branches on
`isMainThread`. Without the `wasm-export` flag no worker is spawned at all;
with it there is always at least one, since a headless render has nowhere
else to go."
(:require
["generic-pool" :as gp]
["node:path" :as path]
["node:process" :as proc]
["node:worker_threads" :as wt]
[app.common.exceptions :as ex]
[app.common.logging :as l]
[app.common.transit :as t]
[app.config :as cf]
[promesa.core :as p]))
(l/set-level! :info)
(defonce pool (atom nil))
(defonce ^:private worker-id (atom 0))
(def ^:private ready-timeout-ms 60000)
(defn- idle-timeout-ms
"How long a render may go silent before the worker is presumed wedged. Reset
on every message, so a long export keeps its worker as long as it keeps
reporting objects; only a thread stuck inside Skia, which reports nothing and
emits no `exit`, runs it out."
[]
(* 1000 (cf/get :wasm-worker-idle-timeout 300)))
(defn- worker-script
[]
(path/resolve (aget (.-argv proc/default) 1)))
(defn- create-worker
[]
(p/create
(fn [resolve reject]
(let [script (worker-script)
id (swap! worker-id inc)
worker (new wt/Worker script)
timer (js/setTimeout
(fn []
(l/error :hint "render worker did not become ready" :worker-id id)
(.terminate ^js worker)
(reject (ex/error :type :internal
:code :worker-not-ready
:hint "render worker did not become ready")))
ready-timeout-ms)]
(unchecked-set worker "__id" id)
(unchecked-set worker "__alive" true)
(.on ^js worker "error"
(fn [cause]
(l/error :hint "render worker error" :worker-id id :cause cause)
(unchecked-set worker "__alive" false)
;; A worker that dies while booting has to fail its own creation;
;; rejecting after `resolve` is a no-op, so this is safe for the
;; errors that arrive once it is already in the pool.
(js/clearTimeout timer)
(reject cause)))
(.on ^js worker "exit"
(fn [code]
(l/info :hint "render worker exited" :worker-id id :code code)
(unchecked-set worker "__alive" false)))
;; Not `.once`: a stray message before the handshake would consume the
;; listener and leave the worker hanging until `ready-timeout-ms`.
(letfn [(on-ready [data]
(when (= "ready" (unchecked-get data "type"))
(js/clearTimeout timer)
(.off ^js worker "message" on-ready)
(l/info :origin "factory" :action "create" :worker-id id)
(resolve worker)))]
(.on ^js worker "message" on-ready))))))
(def ^:private worker-pool-factory
#js {:create create-worker
:destroy (fn [worker]
(l/info :origin "factory" :action "destroy"
:worker-id (unchecked-get worker "__id"))
(.terminate ^js worker))
:validate (fn [worker]
(p/resolved (true? (unchecked-get worker "__alive"))))})
(defn capacity
"How many renders can run at once, and so how many headless jobs the
scheduler may admit. Zero exactly when headless export is off, which is also
when no job is headless, so a headless job always has a worker to wait for."
[]
(if (contains? cf/flags :wasm-export)
;; Clamped rather than rejected: a bad value should not stop the exporter
;; from booting, and a headless render has no other backend to fall back to.
(max 1 (cf/get :wasm-worker-pool-max 2))
0))
(defn init
[]
(let [configured (cf/get :wasm-worker-pool-max 2)
max-workers (capacity)]
(when (and (pos? max-workers) (not= configured max-workers))
(l/warn :hint "wasm-worker-pool-max raised to the minimum of one"
:configured configured))
(if (pos? max-workers)
(let [opts #js {:max max-workers
:min (min max-workers (cf/get :wasm-worker-pool-min 1))
:testOnBorrow true
:evictionRunIntervalMillis 30000
:numTestsPerEvictionRun 2
:idleTimeoutMillis 300000}]
(l/info :hint "initializing render worker pool" :opts opts)
(reset! pool (gp/createPool worker-pool-factory opts)))
(l/info :hint "render worker pool disabled, wasm export is off"))
(p/resolved nil)))
(defn stop
[]
(when-let [instance @pool]
(l/info :hint "finalizing render worker pool")
(reset! pool nil)
(p/do
(.drain ^js instance)
(.clear ^js instance))))
(defn- run-on-worker
"Settles when the worker reports the render finished, failed, or the thread
went away. That last case matters: a terminated worker (how a cancel stops a
render mid-Skia) emits `exit` and never `error`, and a promise left pending
there would keep its pool slot borrowed for the life of the process."
[^js worker params cancel-buffer on-object]
(p/create
(fn [resolve reject]
(let [timer (volatile! nil)]
(letfn [(disarm []
(when-let [t @timer]
(js/clearTimeout t)
(vreset! timer nil)))
(rearm []
(disarm)
(vreset! timer (js/setTimeout
(fn []
(l/error :hint "render worker went silent, terminating"
:worker-id (unchecked-get worker "__id"))
(cleanup)
;; Terminating is what frees the pool slot:
;; the `exit` it raises has no listener left.
(unchecked-set worker "__alive" false)
(.terminate ^js worker)
(reject (ex/error :type :internal
:code :render-timeout
:hint "render worker stopped responding")))
(idle-timeout-ms))))
(cleanup []
(disarm)
(.off worker "message" on-message)
(.off worker "error" on-error)
(.off worker "exit" on-exit))
(on-error [cause]
(cleanup)
(reject cause))
(on-exit [code]
(cleanup)
(reject (ex/error :type :internal
:code :worker-exited
:hint (str "render worker exited with code " code))))
(on-message [data]
(rearm)
(case (unchecked-get data "type")
;; A failure while the main thread handles the object (moving
;; the file, appending to the zip) has to end the render too,
;; or nothing ever settles this promise.
"object" (try
(on-object (t/decode-str (unchecked-get data "payload")))
(catch :default cause
(cleanup)
(reject cause)))
"done" (do (cleanup) (resolve nil))
"error" (do (cleanup)
(reject (ex/error :type :internal
:code (or (some-> (unchecked-get data "code") keyword)
:wasm-render-error)
:hint (unchecked-get data "message"))))
nil))]
(.on worker "message" on-message)
(.once worker "error" on-error)
(.once worker "exit" on-exit)
(rearm)
(.postMessage worker #js {:type "render"
:params (t/encode-str params)
:cancel cancel-buffer}))))))
(defn with-worker
"Acquires one worker for the whole of `f`, a fn of that worker."
[f]
(let [instance @pool]
(->> (p/do (.acquire ^js instance))
(p/mcat (fn [worker]
(->> (p/do (f worker))
(p/fmap (fn [result]
(.release ^js instance worker)
result))
(p/merr (fn [cause]
;; The module may be aborted or mid-write, and
;; a terminated worker cannot be reused.
(-> (p/do (.destroy ^js instance worker))
(p/handle (fn [_ _] (p/rejected cause))))))))))))
(defn render-on
"Renders `params` on an already acquired worker."
[worker params on-object {:keys [cancel-buffer cancelled?]}]
(if (and cancelled? (cancelled?))
(p/rejected (ex/error :type :internal
:code :job-cancelled
:hint "export job was cancelled"))
(run-on-worker worker params cancel-buffer on-object)))
(defn terminate!
[^js worker]
(when worker
(unchecked-set worker "__alive" false)
(.terminate worker)))

452
exporter/src/app/wasm/render.cljs Normal file
View File

@ -0,0 +1,452 @@
;; This Source Code Form is subject to the terms of the Mozilla Public
;; License, v. 2.0. If a copy of the MPL was not distributed with this
;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
;;
;; Copyright (c) KALEIDOS INC Sucursal en España SL
(ns app.wasm.render
"Headless render pipeline: renders exports with the render-wasm Skia pipeline,
with no browser and no WebGL.
Per request: fetch scene (get-page RPC) -> serialize -> provision fonts and
images -> relayout text with the real fonts -> render each object.
This runs inside a render worker (`app.wasm.worker`), one WASM design state
per worker, so the synchronous Skia calls never block the process that serves
HTTP. `app.renderer.wasm` is the main-thread side that drives it.
Moved here verbatim from `app.renderer.wasm`; git reads it as a new file only
because that namespace still exists as the proxy. Reviewable as a rename:
`git show <base>:exporter/src/app/renderer/wasm.cljs | diff -u - <this file>`.
Handles png/jpeg/webp (Skia encodes all three) and pdf; `:svg` stays on the
browser path."
(:require
["node:fs" :as fs]
["undici" :as http]
[app.common.data :as d]
[app.common.exceptions :as ex]
[app.common.fonts :as cfnt]
;; Required for side effects: these register the transit read handlers and
;; deftype impls the `get-page` response is decoded into.
[app.common.geom.matrix]
[app.common.geom.point]
[app.common.geom.rect]
[app.common.logging :as l]
[app.common.transit :as t]
[app.common.types.fills.impl]
[app.common.types.objects-map]
[app.common.types.path.impl]
[app.common.types.shape]
[app.common.types.shape.images :as images]
[app.common.uri :as u]
[app.common.uuid :as uuid]
[app.config :as cf]
[app.util.mime :as mime]
[app.util.shell :as sh]
[app.wasm :as wasm]
[app.wasm.serialize :as serialize]
[cuerdas.core :as str]
[promesa.core :as p]))
;; --- module lifecycle (one shared, lazily-initialized instance)
(defonce ^:private module* (atom nil))
(defn- ensure-module!
[]
(or @module*
(reset! module* (wasm/init!))))
;; --- backend endpoints
;;
;; Every fetch targets the internal endpoint (falling back to public-uri),
;; in a deployment the exporter reaches the backend over the container network
(defn- internal-uri
"Absolute URI for `path` on the internal (backend) endpoint."
[path]
(-> (cf/get-internal-uri)
(u/ensure-path-slash)
(u/join path)
(str)))
(defn- error-detail
"Node's fetch reports every transport failure as a bare `TypeError: fetch
failed`; the actual reason (TLS rejection, DNS, ECONNREFUSED) is buried in a
nested `cause` chain that the logger does not print. Flattens the chain into
one readable string."
[cause]
(->> (iterate (fn [^js e] (unchecked-get e "cause")) cause)
(take-while some?)
(take 5)
(map (fn [^js e]
(let [code (unchecked-get e "code")
msg (or (unchecked-get e "message") (str e))]
(if code (str code ": " msg) msg))))
(str/join " <- ")))
(defn- fetch!
"`undici/fetch` that fails with an ex-info carrying the target uri and the
unwrapped cause chain, so a failed request says what actually went wrong and
against which endpoint."
[uri opts]
(->> (p/do (http/fetch uri opts))
(p/merr (fn [cause]
(p/rejected (ex-info "http fetch failed"
{:uri uri :detail (error-detail cause)}
cause))))))
(defn- explain
"Log-friendly reason for `cause`: the detail `fetch!` already attached, or a
freshly unwrapped chain for anything else (WASM aborts, decode errors)."
[cause]
(or (:detail (ex-data cause))
(error-detail cause)))
(defn- rpc-headers
"Auth headers for backend RPC calls (management key + bearer)."
[token]
#js {"Content-Type" "application/transit+json"
"X-Shared-Key" (str "exporter " cf/management-key)
"Authorization" (str "Bearer " token)})
(defn- asset-headers
"Auth headers for `/assets/*`. Cookie, not Bearer: those endpoints redirect to
a presigned S3/minio URL, and a Bearer header makes S3 400 (\"multiple
authentication types\")."
[token]
#js {"X-Shared-Key" (str "exporter " cf/management-key)
"Cookie" (str "auth-token=" token)})
;; --- shape bundle fetch (backend RPC)
(defn- fetch-objects
"Fetches the exported roots and their children from the backend via the
`get-page` RPC (`:object-id`, as the browser render path does), using the
same auth the exporter uses elsewhere (management key + bearer)."
[{:keys [file-id page-id share-id token objects]}]
(let [headers (rpc-headers token)
root-ids (into #{} (map :id) objects)
body (t/encode-str (cond-> {:file-id file-id
:page-id page-id}
(seq root-ids) (assoc :object-id root-ids)
share-id (assoc :share-id share-id)))
uri (internal-uri "api/rpc/command/get-page")]
(l/dbg :hint "wasm render: get-page"
:uri uri
:file-id (str file-id)
:page-id (str page-id)
:roots (count root-ids))
(->> (fetch! uri #js {:method "POST" :headers headers :body body})
(p/mcat (fn [^js resp]
(if (= 200 (.-status resp))
(.text resp)
(->> (.text resp)
(p/mcat (fn [resp-body]
(l/error :hint "wasm render: get-page failed"
:uri uri
:status (.-status resp)
:body resp-body)
(p/rejected (ex-info "get-page failed"
{:status (.-status resp)
:body resp-body}))))))))
(p/fmap t/decode-str)
(p/fmap :objects))))
;; --- font resolution
;;
;; The text serializer keeps each font's real uuid, so `wasm/fonts-for-shape`
;; reports it. Custom (team) fonts resolve through the file's font variants,
;; google fonts through the shared `app.common.fonts` catalog; builtin
;; fonts through its bundled family + the frontend's static `/fonts/`.
(defn- fetch-font-variants
"Team (custom) font variants for the file, or nil — a failure here degrades
to fallback fonts, it does not fail the export."
[{:keys [file-id share-id token]}]
(let [headers (rpc-headers token)
body (t/encode-str (cond-> {:file-id file-id}
share-id (assoc :share-id share-id)))
uri (internal-uri "api/rpc/command/get-font-variants")]
(->> (fetch! uri #js {:method "POST" :headers headers :body body})
(p/mcat (fn [^js resp]
(if (= 200 (.-status resp))
(.text resp)
(p/resolved nil))))
(p/fmap (fn [s] (when s (t/decode-str s))))
(p/merr (fn [cause]
(l/warn :hint "wasm render: get-font-variants failed"
:uri uri :detail (explain cause) :cause cause)
(p/resolved nil))))))
(defn- fetch-ttf-bytes
"Downloads a TTF, returning a promise of an ArrayBuffer (or nil). A failure
here degrades to fallback fonts, it does not fail the export."
([uri] (fetch-ttf-bytes uri #js {:method "GET"}))
([uri opts]
(->> (fetch! uri opts)
(p/mcat (fn [^js resp]
(if (= 200 (.-status resp))
(.arrayBuffer resp)
(p/resolved nil))))
(p/merr (fn [cause]
(l/warn :hint "wasm render: font fetch failed"
:uri uri :detail (explain cause) :cause cause)
(p/resolved nil))))))
;; TTF bytes cached for the process lifetime, keyed by whatever identifies the
;; variant (a gfont id+weight+style, a builtin file name).
(defonce ^:private font-bytes* (atom {}))
(defn- cached-ttf-bytes
[cache-key fetch-fn]
(if-let [bytes (get @font-bytes* cache-key)]
(p/resolved bytes)
(->> (fetch-fn)
(p/fmap (fn [buf]
(when buf (swap! font-bytes* assoc cache-key buf))
buf)))))
(defn- fetch-asset-bytes
[asset-id {:keys [token]}]
(fetch-ttf-bytes (internal-uri (str "assets/by-id/" asset-id))
#js {:method "GET" :headers (asset-headers token)}))
(defn- fetch-gfont-bytes
[ttf-url]
(fetch-ttf-bytes (cfnt/gstatic->proxy-url ttf-url (internal-uri "internal/gfonts/font"))))
(defn- fetch-builtin-font-bytes
[ttf-file]
(cached-ttf-bytes ttf-file #(fetch-ttf-bytes (internal-uri (str "fonts/" ttf-file)))))
(defn- make-resolve-font
"Builds a `resolve-font` fn (family map -> promise of TTF bytes). Custom
variants first, matching uuid+weight+style then degrading to uuid+weight then
uuid; the bundled fonts for `uuid/zero`, which is what `font-id->uuid` maps
every builtin family to; google catalog otherwise."
[variants params]
(fn [{:keys [id weight style]}]
(let [font-uuid (uuid/from-unsigned-parts (aget id 0) (aget id 1) (aget id 2) (aget id 3))
style-str (if (zero? style) "normal" "italic")
variant (or (d/seek (fn [v] (and (= (:font-id v) font-uuid)
(= (:font-weight v) weight)
(= (name (:font-style v)) style-str)))
variants)
(d/seek (fn [v] (and (= (:font-id v) font-uuid)
(= (:font-weight v) weight)))
variants)
(d/seek (fn [v] (= (:font-id v) font-uuid)) variants))]
(cond
(:ttf-file-id variant)
(fetch-asset-bytes (:ttf-file-id variant) params)
(= uuid/zero font-uuid)
(fetch-builtin-font-bytes (cfnt/resolve-ttf-file weight style))
:else
(if-let [gurl (cfnt/resolve-ttf-url font-uuid weight style)]
(fetch-gfont-bytes gurl)
(p/resolved nil))))))
;; --- fallback fonts (emoji + per-script noto fonts)
;;
;; Emoji and non-latin scripts render through fallback families, not through
;; any span's font family, so `wasm/fonts-for-shape` never reports them and the
;; provisioning above never uploads them. Must run per request, since
;; `clear-fonts!` empties the store; the TTF bytes stay cached per process.
(defn- scene-fallback-fonts
"Fallback font descriptors needed by the scene's text. Deduped because
several languages map to one noto family and provisioning is concurrent —
otherwise they all miss the byte cache at once and refetch the same TTF."
[scene]
(let [texts (for [shape (vals scene)
:when (= :text (:type shape))
node (or (some->> (:content shape) (tree-seq :children :children)) [])
:let [text (:text node)]
:when (string? text)]
text)
emoji? (boolean (some cfnt/contains-emoji? texts))
langs (reduce cfnt/collect-used-languages #{} texts)]
(distinct
(cond-> (cfnt/add-noto-fonts [] langs)
emoji? (cfnt/add-emoji-font)))))
(defn- fetch-fallback-font-bytes
"Downloads one fallback font's TTF. Cached by the whole variant, not just
`font-id`: `resolve-ttf-url` picks a different TTF per weight/style, so a
font-id-only key would serve the first downloaded variant for every other one."
[{:keys [font-id weight style]}]
(if-let [ttf-url (some-> (cfnt/gfont-id->uuid font-id) (cfnt/resolve-ttf-url weight style))]
(cached-ttf-bytes [font-id weight style] #(fetch-gfont-bytes ttf-url))
(p/resolved nil)))
(defn- provision-fallback-fonts!
[scene]
(->> (scene-fallback-fonts scene)
(map (fn [{:keys [font-id weight style is-emoji is-fallback] :as font}]
(if-let [font-uuid (cfnt/gfont-id->uuid font-id)]
(->> (fetch-fallback-font-bytes font)
(p/fmap (fn [buf]
(if buf
(wasm/store-font! {:id (uuid/get-u32 font-uuid)
:weight weight
:style style
:emoji? (boolean is-emoji)
:fallback? (boolean is-fallback)}
buf)
(l/warn :hint "wasm render: fallback font unavailable"
:font-id font-id)))))
(p/resolved nil))))
(p/all)))
;; --- image resolution
;;
;; Image fills reference file-media ids; the encoded bytes go straight to
;; `_store_image` (Skia decodes, no WebGL), keyed by media uuid so this happens
;; once per request rather than per rendered object.
(defn- fetch-file-media-bytes
"Downloads an image fill's encoded bytes by file-media id."
[media-id {:keys [token]}]
(let [headers (asset-headers token)
uri (internal-uri (str "assets/by-file-media-id/" media-id))]
(->> (fetch! uri #js {:method "GET" :headers headers})
(p/mcat (fn [^js resp]
(if (= 200 (.-status resp))
(.arrayBuffer resp)
(do
(l/warn :hint "wasm render: image fetch non-200"
:media-id (str media-id)
:uri uri
:status (.-status resp))
(p/resolved nil)))))
(p/merr (fn [cause]
(l/warn :hint "wasm render: image fetch failed"
:media-id (str media-id) :uri uri
:detail (explain cause) :cause cause)
(p/resolved nil))))))
(defn- provision-images!
"Fetches and stores every image the scene references (shape, stroke and
text-span fills, enumerated by `app.common.types.shape.images`). Unlike fonts,
the image store is not reset per request, so already-held images are skipped
and repeated exports of a file reuse them."
[scene params]
(let [all-ids (images/scene-image-ids scene)
new-ids (remove wasm/image-cached? all-ids)]
(l/dbg :hint "wasm render: provisioning images"
:total (count all-ids)
:cached (- (count all-ids) (count new-ids)))
(->> new-ids
(map (fn [image-id]
(->> (fetch-file-media-bytes image-id params)
(p/fmap (fn [buf]
(if buf
(do
(l/dbg :hint "wasm render: image stored"
:media-id (str image-id)
:bytes (.-byteLength ^js buf))
(wasm/store-image! image-id buf))
(l/warn :hint "wasm render: image unavailable"
:media-id (str image-id))))))))
(p/all))))
(defn- relayout-text!
"Recomputes layout for every text shape, once the real fonts are provisioned
(serialize-time layout used the fallback)."
[scene]
(doseq [shape (vals scene)
:when (= :text (:type shape))]
(wasm/update-text-layout! (:id shape))))
;; --- render
(defn- check-cancelled!
"Cancellation is cooperative: a render already inside Skia cannot be
interrupted, so the flag is only observed between objects. Killing a job
mid-object is the caller's job (terminating the worker)."
[{:keys [cancelled?] :as _params}]
(when (and cancelled? (cancelled?))
(ex/raise :type :internal
:code :job-cancelled
:hint "export job was cancelled")))
(defn- render-object-bytes
[type id scale]
(if (= :pdf type)
(let [bytes (wasm/render-shape-pdf id scale)]
(l/dbg :hint "PDF generated via Skia (render-wasm headless)"
:object-id (str id)
:backend "skia-wasm"
:bytes (.-length bytes))
bytes)
(wasm/render-shape-raster id scale type)))
(defn- render*
[{:keys [scale type objects] :as params} on-object]
(l/dbg :hint "wasm render: start"
:type type
:scale scale
:objects (count objects)
:file-id (str (:file-id params))
:page-id (str (:page-id params)))
(->> (ensure-module!)
(p/mcat (fn [_] (fetch-objects params)))
(p/mcat (fn [scene]
(l/dbg :hint "wasm render: scene fetched" :shapes (count scene))
(serialize/serialize-scene! scene)
(l/dbg :hint "wasm render: scene serialized")
;; So fonts from a previous request don't leak into this one.
(wasm/clear-fonts!)
(->> (p/all [(fetch-font-variants params)
(provision-images! scene params)
(provision-fallback-fonts! scene)])
(p/mcat
(fn [[variants _]]
(let [resolve-font (make-resolve-font (or variants []) params)]
;; Before rendering, so the relayout below sees real
;; font metrics. Deduped across objects: shapes
;; sharing one family download its TTF once.
(wasm/provision-fonts! (map :id objects) resolve-font))))
(p/mcat
(fn [_]
(relayout-text! scene)
(p/run
(fn [{:keys [id] :as object}]
(check-cancelled! params)
(let [bytes (render-object-bytes type id scale)
path (sh/tempfile :prefix "penpot.tmp.wasm."
:suffix (mime/get-extension type))]
(l/dbg :hint "wasm render: object rendered"
:object-id (str id) :bytes (.-length bytes))
(fs/writeFileSync path bytes)
;; `on-object` returns a plain value (zip append) or
;; a promise (single export's file move); `p/do`
;; normalizes both to a thenable.
(p/do (on-object (assoc object :path path)))))
objects))))))
(p/fmap (fn [result]
;; After the request, never mid-render, so an image can't
;; disappear under a running export.
(let [evicted (wasm/evict-images! (cf/get :wasm-worker-image-cache-size wasm/image-cache-size))]
(when (pos? evicted)
(l/info :hint "wasm render: evicted cached images" :count evicted)))
result))
(p/merr (fn [cause]
(l/error :hint "wasm render: failed"
:detail (explain cause)
:internal-uri (str (cf/get-internal-uri))
:cause cause)
;; A panic can leave the mem buffer allocated or the instance
;; aborted; drop it so the next request rebuilds a fresh one.
(reset! module* nil)
(p/rejected cause)))))
(defn render
"Public entry. Renders every object of `params`, calling `on-object` with
`{:id :filename :path ...}` as each one is written out."
[params on-object]
(render* params on-object))

71
exporter/src/app/wasm/worker.cljs Normal file
View File

@ -0,0 +1,71 @@
;; This Source Code Form is subject to the terms of the Mozilla Public
;; License, v. 2.0. If a copy of the MPL was not distributed with this
;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
;;
;; Copyright (c) KALEIDOS INC Sucursal en España SL
(ns app.wasm.worker
"Render worker entry point.
Owns one render-wasm module and renders one export request at a time. The
Skia calls are synchronous, so running them here is what lets several exports
progress at once: the main thread keeps serving HTTP, zipping and uploading
while this thread is blocked inside a render.
Messages in: {type: \"render\", params: <transit>, cancel: SharedArrayBuffer}
Messages out: {type: \"ready\"}
{type: \"object\", payload: <transit>} one per rendered object
{type: \"done\"} | {type: \"error\", message, code}"
(:require
["node:worker_threads" :as wt]
[app.common.logging :as l]
[app.common.transit :as t]
[app.wasm.render :as render]
[promesa.core :as p]))
(defn- post!
[message]
(.postMessage ^js wt/parentPort message))
(defn- cancelled-fn
[buffer]
(if (some? buffer)
(let [signal (js/Int32Array. buffer)]
(fn [] (pos? (js/Atomics.load signal 0))))
(constantly false)))
(defn- handle-render
[data]
(let [params (-> (unchecked-get data "params")
(t/decode-str)
(assoc :cancelled? (cancelled-fn (unchecked-get data "cancel"))))]
(->> (render/render params
(fn [object]
(post! #js {:type "object" :payload (t/encode-str object)})))
(p/fmap (fn [_] (post! #js {:type "done"})))
(p/merr (fn [cause]
(l/warn :hint "render worker: request failed" :cause cause)
(post! #js {:type "error"
:message (or (ex-message cause) (str cause))
:code (some-> cause ex-data :code name)})
(p/resolved nil))))))
(defn- on-message
[data]
(case (unchecked-get data "type")
"render" (handle-render data)
(l/warn :hint "render worker: unknown message" :type (unchecked-get data "type"))))
(defonce ^:private listening
;; `defonce` survives a hot reload, so a reload does not stack a second
;; listener on the port. The indirection through the var keeps the reloaded
;; `on-message` in play instead of pinning the one captured at boot.
(delay
(.on ^js wt/parentPort "message" (fn [data] (on-message data)))
true))
(defn main
[& _]
@listening
(post! #js {:type "ready"})
(l/info :hint "render worker ready"))

31
exporter/test/exporter_tests/export_shapes_test.cljs Normal file
View File

@ -0,0 +1,31 @@
;; This Source Code Form is subject to the terms of the Mozilla Public
;; License, v. 2.0. If a copy of the MPL was not distributed with this
;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
;;
;; Copyright (c) KALEIDOS INC Sucursal en España SL
(ns exporter-tests.export-shapes-test
"Chunking of the browser backend."
(:require
[app.common.uuid :as uuid]
[app.handlers.export-shapes :as export-shapes]
[cljs.test :as t :include-macros true]))
(defn- exports
[n type scale]
(let [file-id (uuid/next)
page-id (uuid/next)]
(mapv (fn [i]
{:file-id file-id
:page-id page-id
:object-id (uuid/next)
:name (str "shape-" i)
:suffix ""
:scale scale
:type type})
(range n))))
(t/deftest browser-exports-are-chunked
(let [parts (export-shapes/prepare-exports (exports 120 :png 1) "token" false)]
(t/is (= 3 (count parts)))
(t/is (= [50 50 20] (mapv (comp count :objects) parts)))))

86
exporter/test/exporter_tests/jobs_test.cljs Normal file
View File

@ -0,0 +1,86 @@
;; This Source Code Form is subject to the terms of the Mozilla Public
;; License, v. 2.0. If a copy of the MPL was not distributed with this
;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
;;
;; Copyright (c) KALEIDOS INC Sucursal en España SL
(ns exporter-tests.jobs-test
"Job state machine. Runs without redis: a store write with no connection is
reported and swallowed, so only the in-process record is exercised."
(:require
[app.common.uuid :as uuid]
[app.jobs :as jobs]
[cljs.test :as t :include-macros true]
[promesa.core :as p]))
(defn- create!
[]
(jobs/create! {:profile-id (uuid/next)
:cmd :export-shapes
:backend "wasm"
:total 10
:name "test"
:resource-id (uuid/next)}
(constantly (p/resolved nil))))
(t/deftest progress-does-not-resurrect-a-finished-job
(t/testing "a render reporting after the export failed cannot undo the failure"
(t/async done
(p/let [job (create!)
_ (jobs/start! job)
_ (jobs/fail! (jobs/lookup (:id job)) (ex-info "boom" {}))
;; `job` is the snapshot handed to the work when it started, which
;; is what a straggling render still holds.
_ (jobs/progress! job 7)]
(let [current (jobs/lookup (:id job))]
(t/is (= "error" (:state current)))
(t/is (= "boom" (:error current)))
(t/is (not= 7 (:done current))))
(jobs/release! (:id job))
(done)))))
(t/deftest first-terminal-state-wins
(t/testing "a failure arriving after a cancellation leaves the job cancelled"
(t/async done
(p/let [job (create!)
_ (jobs/start! job)
_ (jobs/cancel! (:id job))
_ (jobs/fail! job (ex-info "too late" {}))]
(let [current (jobs/lookup (:id job))]
(t/is (= "cancelled" (:state current)))
(t/is (nil? (:error current))))
(jobs/release! (:id job))
(done)))))
(t/deftest cancel-is-recorded-before-the-callbacks-run
(t/testing "a queued job dropped by its own cancel callback still ends cancelled"
(t/async done
(let [seen (atom ::not-called)]
(p/let [job (create!)
;; What `scheduler/drop-queued!` does: it takes the job off the
;; queue and releases it. Anything the lifecycle wrote after
;; the callbacks ran would be dropped on the floor, so by the
;; time one is called the record has to be terminal already.
_ (jobs/on-cancel (:id job)
(fn []
(reset! seen (:state (jobs/lookup (:id job))))
(jobs/release! (:id job))))
_ (jobs/cancel! (:id job))]
(t/is (= "cancelled" @seen))
(t/is (nil? (jobs/lookup (:id job))))
(done))))))
(t/deftest writes-stop-once-the-job-is-released
(t/testing "a late write for a job the scheduler already settled is dropped"
(t/async done
(p/let [job (create!)
_ (jobs/start! job)
_ (jobs/complete! (jobs/lookup (:id job)) {:uri "http://example/x"
:filename "x.zip"
:mtype "application/zip"})
ended (jobs/lookup (:id job))
_ (jobs/release! (:id job))
_ (jobs/progress! job 3)]
(t/is (= "ended" (:state ended)))
(t/is (nil? (jobs/lookup (:id job))))
(done)))))

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

@ -10,15 +10,23 @@
[cljs.test :as t]
[clojure.string :as str]
[clojure.tools.cli :refer [parse-opts]]
[exporter-tests.export-shapes-test]
[exporter-tests.jobs-test]
[exporter-tests.renderer-svg-test]
[exporter-tests.scheduler-test]
[exporter-tests.shell-test]
[exporter-tests.wasm-pool-test]
[goog.object :as gobj]))
(enable-console-print!)
(def test-namespaces
['exporter-tests.renderer-svg-test
'exporter-tests.shell-test])
['exporter-tests.export-shapes-test
'exporter-tests.jobs-test
'exporter-tests.renderer-svg-test
'exporter-tests.scheduler-test
'exporter-tests.shell-test
'exporter-tests.wasm-pool-test])
(assert (every? find-ns-obj test-namespaces)
"test-namespaces contains a namespace that isn't required in runner.cljs")

81
exporter/test/exporter_tests/scheduler_test.cljs Normal file
View File

@ -0,0 +1,81 @@
;; This Source Code Form is subject to the terms of the Mozilla Public
;; License, v. 2.0. If a copy of the MPL was not distributed with this
;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
;;
;; Copyright (c) KALEIDOS INC Sucursal en España SL
(ns exporter-tests.scheduler-test
"Admission control. A headless job leases one render worker for its whole run,
so no more of them may start than there are workers."
(:require
[app.common.uuid :as uuid]
[app.jobs :as jobs]
[app.jobs.scheduler :as scheduler]
[app.wasm.pool :as pool]
[cljs.test :as t :include-macros true]
[promesa.core :as p]))
(defn- pin-capacity!
"Fixes the worker count for the test. Returns the thunk that puts it back;
`with-redefs` cannot be used here, since the scheduler keeps admitting jobs
after the body of the test has returned."
[n]
(let [original pool/capacity]
(set! pool/capacity (constantly n))
(fn [] (set! pool/capacity original))))
(defn- create!
[backend run-fn]
(jobs/create! {:profile-id (uuid/next)
:cmd :export-shapes
:backend backend
:total 1
:name "test"
:resource-id (uuid/next)}
run-fn))
(defn- gate
"A promise and the fn that settles it, standing in for a render in flight."
[]
(let [resolve* (volatile! nil)
pending (p/create (fn [resolve _] (vreset! resolve* resolve)))]
[pending (fn [] (@resolve* nil))]))
(t/deftest headless-jobs-wait-for-a-render-worker
(t/testing "with one worker, the second headless job stays queued until the first ends"
(t/async done
(let [restore! (pin-capacity! 1)
[render open] (gate)
started (atom [])]
(p/let [job1 (create! "wasm" (fn [_] (swap! started conj :one) render))
job2 (create! "wasm" (fn [_] (swap! started conj :two) (p/resolved nil)))]
(let [p1 (scheduler/submit! job1)
p2 (scheduler/submit! job2)]
(p/do
(p/delay 10)
(t/is (= [:one] @started))
(t/is (= "running" (:state (jobs/lookup (:id job1)))))
(t/is (= "queued" (:state (jobs/lookup (:id job2)))))
(open)
(p/all [p1 p2])
(t/is (= [:one :two] @started))
(restore!)
(done))))))))
(t/deftest a-browser-job-is-not-held-back-by-the-worker-pool
(t/testing "the headless cap applies to headless jobs only"
(t/async done
(let [restore! (pin-capacity! 1)
[render open] (gate)
started (atom [])]
(p/let [job1 (create! "wasm" (fn [_] (swap! started conj :wasm) render))
job2 (create! "browser" (fn [_] (swap! started conj :browser) (p/resolved nil)))]
(let [p1 (scheduler/submit! job1)
p2 (scheduler/submit! job2)]
(p/do
(p/delay 10)
(t/is (= [:wasm :browser] @started))
(open)
(p/all [p1 p2])
(restore!)
(done))))))))

46
exporter/test/exporter_tests/wasm_pool_test.cljs Normal file
View File

@ -0,0 +1,46 @@
;; This Source Code Form is subject to the terms of the Mozilla Public
;; License, v. 2.0. If a copy of the MPL was not distributed with this
;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
;;
;; Copyright (c) KALEIDOS INC Sucursal en España SL
(ns exporter-tests.wasm-pool-test
"Worker leasing, against a stub pool: `with-worker` must give the worker back
however its body ends."
(:require
[app.wasm.pool :as pool]
[cljs.test :as t :include-macros true]
[promesa.core :as p]))
(defn- stub-pool!
"Installs a pool whose acquire/release/destroy only count calls."
[]
(let [calls (atom {:acquired 0 :released 0 :destroyed 0})]
(reset! pool/pool
#js {:acquire (fn [] (swap! calls update :acquired inc) (p/resolved ::worker))
:release (fn [_] (swap! calls update :released inc) (p/resolved nil))
:destroy (fn [_] (swap! calls update :destroyed inc) (p/resolved nil))})
calls))
(t/deftest releases-the-worker-when-the-body-succeeds
(t/async done
(let [calls (stub-pool!)]
(p/let [result (pool/with-worker (fn [_] (p/resolved :ok)))]
(t/is (= :ok result))
(t/is (= 1 (:acquired @calls)))
(t/is (= 1 (:released @calls)))
(t/is (= 0 (:destroyed @calls)))
(reset! pool/pool nil)
(done)))))
(t/deftest gives-the-worker-back-when-the-body-throws-synchronously
(t/testing "a raise out of the scope body must not leave the worker borrowed"
(t/async done
(let [calls (stub-pool!)]
(->> (pool/with-worker (fn [_] (throw (ex-info "cancelled" {}))))
(p/hmap (fn [_ cause]
(t/is (some? cause))
(t/is (= 1 (:acquired @calls)))
(t/is (= 1 (+ (:released @calls) (:destroyed @calls))))
(reset! pool/pool nil)
(done))))))))

153
frontend/src/app/main/data/exports/assets.cljs
View File

@ -8,6 +8,7 @@
(:require
[app.common.time :as ct]
[app.common.uuid :as uuid]
[app.config :as cf]
[app.main.data.event :as ev]
[app.main.data.exports.wasm :as wasm.exports]
[app.main.data.helpers :as dsh]
@ -141,35 +142,57 @@
:name page-name}))))))
(defn- initialize-export-status
[exports cmd resource]
"`job` is only present on the job API path; without it the widget counts the
exports the client submitted, exactly as it always has."
[exports cmd resource {:keys [job-id total status backend] :as job}]
(ptk/reify ::initialize-export-status
ptk/UpdateEvent
(update [_ state]
(assoc state :export {:in-progress true
:resource-id (:id resource)
:healthy? true
:error false
:progress 0
:widget-visible true
:detail-visible true
:exports exports
:last-update (ct/now)
:cmd cmd}))))
(assoc state :export (cond-> {:in-progress true
:resource-id (:id resource)
:healthy? true
:error false
:progress 0
:widget-visible true
:detail-visible true
:exports exports
:last-update (ct/now)
:cmd cmd}
(some? job)
(assoc :job-id job-id
:total total
:status status
:backend backend))))))
(defn- update-export-status
[{:keys [done status resource-uri filename mtype] :as data}]
[{:keys [done total status resource-uri filename mtype] :as data}]
(ptk/reify ::update-export-status
ptk/UpdateEvent
(update [_ state]
(let [time-diff (ct/diff-ms (get-in state [:export :last-update]) (ct/now))
healthy? (< time-diff 6000)]
healthy? (< time-diff 6000)
;; The legacy path has no server-side figures to track; it keeps
;; reporting progress over the client's own list.
job? (some? (get-in state [:export :job-id]))]
(cond-> state
job?
(update :export assoc :status status)
(and job? (some? total))
(update :export assoc :total total)
(= status "running")
(update :export assoc :progress done :last-update (ct/now) :healthy? healthy?)
(= status "error")
(update :export assoc :in-progress false :error (:cause data) :last-update (ct/now) :healthy? healthy?)
(= status "cancelling")
(update :export assoc :last-update (ct/now) :healthy? healthy?)
(= status "cancelled")
(update :export assoc :in-progress false :last-update (ct/now) :healthy? healthy?)
(= status "ended")
(update :export assoc :in-progress false :last-update (ct/now) :healthy? healthy?))))
@ -178,17 +201,78 @@
(when (= status "ended")
(dom/trigger-download-uri filename mtype resource-uri)))))
;; The exporter is at capacity. Not a crash: the widget says so and the user
;; retries, instead of the generic error dialog.
(def ^:private saturation-codes #{:queue-full})
(defn- export-failed
"Reports a failure that happened before the export ever started, so the widget
settles instead of waiting for progress that will never arrive."
[exports cmd cause]
(ptk/reify ::export-failed
ptk/UpdateEvent
(update [_ state]
(assoc state :export {:in-progress false
:widget-visible true
:detail-visible true
:healthy? true
:progress 0
:total (count exports)
:exports exports
:cmd cmd
:error (or (ex-message cause) true)
:error-code (:code (ex-data cause))
:last-update (ct/now)}))))
(defn cancel-export
"Stops the running export. Only reachable on the job API path, where the
exporter can actually abort the work.
The widget settles from here rather than from the job's `cancelled` message:
the outcome is known once the request returns, and waiting on a round trip
through redis and the websocket would leave it stuck whenever that message is
missed."
[]
(ptk/reify ::cancel-export
ptk/WatchEvent
(watch [_ state _]
(when-let [job-id (get-in state [:export :job-id])]
(let [resource-id (get-in state [:export :resource-id])
settle (rx/concat
(rx/of (update-export-status {:status "cancelled"}))
(->> (rx/of (clear-export-state resource-id))
(rx/delay default-timeout)))]
(rx/concat
;; Stopping is not instantaneous: the request has to reach the
;; exporter and the work has to unwind.
(rx/of (update-export-status {:status "cancelling"}))
(->> (rp/cmd! :cancel-export-job {:job-id job-id})
(rx/mapcat (fn [_] settle))
;; Already finished, or the exporter is gone; either way
;; there is nothing left to stop.
(rx/catch (fn [_] settle)))))))))
;; TODO: Remove once we support WASM SVG export
(def ^:private wasm-export-types #{:jpeg :webp :png :pdf})
(defn- wasm-export-enabled?
"WASM export is available when the `wasm-export/v1` feature is active AND
render-wasm is active for the current file. When render-wasm is inactive its
shape tree isn't loaded, so a client-side WASM render would crash."
shape tree isn't loaded, so a client-side WASM render would crash.
This governs the client-side render only; it says nothing about the exporter."
[state]
(and (features/active-feature? state "wasm-export/v1")
(features/active-feature? state "render-wasm/v1")))
(defn- wasm-export-available?
"Whether the *exporter* renders with render-wasm. Its `enable-wasm-export`
flag has to be on too, otherwise the browser backend does the work and the
job API would promise capabilities the server does not have."
[state]
(and (wasm-export-enabled? state)
(contains? cf/flags :wasm-export)))
(defn- use-wasm-export?
"Whether to take the client-side WASM export path for `export`."
[state export]
@ -223,7 +307,7 @@
:profile-id profile-id
:cmd :export-shapes
:wait true
:is-wasm (wasm-export-enabled? state)})]
:is-wasm (wasm-export-available? state)})]
(rx/concat
(dwp/force-persist-and-wait 400)
@ -252,7 +336,7 @@
:cmd cmd
:profile-id profile-id
:force-multiple true
:is-wasm (wasm-export-enabled? state)}
:is-wasm (wasm-export-available? state)}
(some? name)
(assoc :name name))
@ -266,7 +350,8 @@
stopper
(rx/filter #(or (= "ended" (:status %))
(= "error" (:status %)))
(= "error" (:status %))
(= "cancelled" (:status %)))
progress-stream)]
(swap! st/ongoing-tasks conj :export)
@ -276,11 +361,30 @@
(rx/of ::dwp/force-persist)
;; Launch the exportation process and stores the resource id
;; locally.
(->> (rp/cmd! :export params)
(rx/map (fn [{:keys [id] :as resource}]
(vreset! resource-id id)
(initialize-export-status exports cmd resource))))
;; locally. With wasm export active the job API is used instead: it
;; answers with the exporter's own object count and gives a handle
;; to cancel.
(->> (if (wasm-export-available? state)
(->> (rp/cmd! :create-export-job params)
(rx/map (fn [{job-id :id :keys [total] :as job}]
(vreset! resource-id (:resource-id job))
(initialize-export-status exports cmd
{:id (:resource-id job)}
{:job-id job-id
:total total
:status (:state job)
:backend (:backend job)}))))
(->> (rp/cmd! :export params)
(rx/map (fn [{:keys [id] :as resource}]
(vreset! resource-id id)
(initialize-export-status exports cmd resource nil)))))
(rx/catch (fn [cause]
;; Saturation is an answer, not a fault.
(if (contains? saturation-codes (:code (ex-data cause)))
(rx/of (export-failed exports cmd cause))
(rx/concat
(rx/of (export-failed exports cmd cause))
(rx/throw cause))))))
;; We proceed to update the export state with incoming
;; progress updates. We delay the stopper for give some time
@ -297,7 +401,8 @@
;; for ensure that after some security time, the stream is
;; completely closed.
(->> progress-stream
(rx/filter #(= "ended" (:status %)))
(rx/filter #(or (= "ended" (:status %))
(= "cancelled" (:status %))))
(rx/take 1)
(rx/delay default-timeout)
(rx/map #(clear-export-state @resource-id))
@ -316,7 +421,7 @@
(watch [_ state _]
(let [params (select-keys (:export state) [:exports :cmd])]
(when (seq params)
(rx/of (request-multiple-export params)))))))
(rx/of (request-export params)))))))
(defn export-shapes-event
[exports origin]

22
frontend/src/app/main/repo.cljs
View File

@ -276,6 +276,28 @@
(let [default {:wait false :blob? false}]
(send-export (merge default params))))
(defmethod cmd! :create-export-job
[_ params]
(->> (http/send! {:method :post
:uri (u/join cf/public-uri "api/export/jobs")
:body (http/transit-data params)
:headers {"x-external-session-id" (cf/external-session-id)
"x-event-origin" (::ev/origin (meta params))}
:credentials "include"
:response-type :text})
(rx/map http/conditional-decode-transit)
(rx/mapcat handle-response)))
(defmethod cmd! :cancel-export-job
[_ {:keys [job-id]}]
(->> (http/send! {:method :delete
:uri (u/join cf/public-uri "api/export/jobs/" (str job-id))
:headers {"x-external-session-id" (cf/external-session-id)}
:credentials "include"
:response-type :text})
(rx/map http/conditional-decode-transit)
(rx/mapcat handle-response)))
(defn- multipart-upload
[id params]
(->> (http/send! {:method :post

41
frontend/src/app/main/ui/exports/assets.cljs
View File

@ -218,11 +218,25 @@
theme (or (:theme profile) theme/default)
is-default-theme? (= theme/default theme)
error? (:error state)
;; The exporter is at capacity: worth its own wording, so the user
;; knows retrying later is the thing to do.
busy? (= :queue-full (:error-code state))
healthy? (:healthy? state)
detail-visible? (:detail-visible state)
widget-visible? (:widget-visible state)
progress (:progress state)
items (:exports state)
job-id (:job-id state)
status (:status state)
queued? (and (some? job-id) (= "queued" status))
cancelling? (and (some? job-id) (= "cancelling" status))
cancelled? (and (some? job-id) (= "cancelled" status))
;; Only the wasm backend can actually stop: a browser render holds its
;; pool slot until playwright gives up.
cancellable? (and (some? job-id)
(= "wasm" (:backend state))
(:in-progress state)
(not cancelling?))
total (or (:total state) (count items))
complete? (= progress total)
circ (* 2 Math/PI 12)
@ -236,6 +250,8 @@
color
(cond
error? clr/new-danger
(or cancelling?
cancelled?) clr/new-warning
healthy? (if is-default-theme?
clr/new-primary
clr/new-primary-light)
@ -248,11 +264,20 @@
title
(cond
busy? (tr "workspace.options.exporting-busy")
error? (tr "workspace.options.exporting-object-error")
cancelling? (tr "workspace.options.exporting-cancelling")
cancelled? (tr "workspace.options.exporting-cancelled")
queued? (tr "workspace.options.exporting-queued")
complete? (tr "workspace.options.exporting-complete")
healthy? (tr "workspace.options.exporting-object")
(not healthy?) (tr "workspace.options.exporting-object-slow"))
cancel-export
(mf/use-fn
(fn []
(st/emit! (de/cancel-export))))
retry-last-operation
(mf/use-fn
(fn []
@ -294,11 +319,25 @@
[:div {:class (stl/css :export-progress-title)}
[:div {:class (stl/css :title-text)} title]
(if error?
(cond
error?
[:button {:class (stl/css :retry-btn)
:on-click retry-last-operation}
(tr "workspace.options.retry")]
cancellable?
[:*
[:button {:class (stl/css :retry-btn)
:on-click cancel-export}
(tr "workspace.options.cancel-export")]
[:span {:class (stl/css :progress)}
(dm/str progress " / " total)]]
;; A counter for work that is being abandoned says nothing useful.
(or cancelling? cancelled?)
nil
:else
[:span {:class (stl/css :progress)}
(dm/str progress " / " total)])]

20
frontend/translations/en.po
View File

@ -7775,10 +7775,30 @@ msgstr "Remove export"
msgid "workspace.options.export.suffix"
msgstr "Suffix"
#: src/app/main/ui/exports/assets.cljs:325
msgid "workspace.options.cancel-export"
msgstr "Cancel"
#: src/app/main/ui/exports/assets.cljs:252
msgid "workspace.options.exporting-complete"
msgstr "Export complete"
#: src/app/main/ui/exports/assets.cljs:259
msgid "workspace.options.exporting-cancelled"
msgstr "Export cancelled"
#: src/app/main/ui/exports/assets.cljs:258
msgid "workspace.options.exporting-cancelling"
msgstr "Cancelling..."
#: src/app/main/ui/exports/assets.cljs:261
msgid "workspace.options.exporting-queued"
msgstr "Waiting..."
#: src/app/main/ui/exports/assets.cljs:256
msgid "workspace.options.exporting-busy"
msgstr "Export service is busy, please try again later"
#: src/app/main/ui/exports/assets.cljs:171, src/app/main/ui/exports/assets.cljs:253, src/app/main/ui/inspect/exports.cljs:216, src/app/main/ui/workspace/sidebar/options/menus/exports.cljs:273
msgid "workspace.options.exporting-object"
msgstr "Exporting…"

20
frontend/translations/es.po
View File

@ -7564,10 +7564,30 @@ msgstr "Eliminar exportación"
msgid "workspace.options.export.suffix"
msgstr "Sufijo"
#: src/app/main/ui/exports/assets.cljs:325
msgid "workspace.options.cancel-export"
msgstr "Cancelar"
#: src/app/main/ui/exports/assets.cljs:252
msgid "workspace.options.exporting-complete"
msgstr "Exportación completa"
#: src/app/main/ui/exports/assets.cljs:259
msgid "workspace.options.exporting-cancelled"
msgstr "Exportación cancelada"
#: src/app/main/ui/exports/assets.cljs:258
msgid "workspace.options.exporting-cancelling"
msgstr "Cancelando..."
#: src/app/main/ui/exports/assets.cljs:261
msgid "workspace.options.exporting-queued"
msgstr "Esperando..."
#: src/app/main/ui/exports/assets.cljs:256
msgid "workspace.options.exporting-busy"
msgstr "La cola de exportación está llena, inténtalo de nuevo en unos momentos"
#: src/app/main/ui/exports/assets.cljs:171, src/app/main/ui/exports/assets.cljs:253, src/app/main/ui/inspect/exports.cljs:216, src/app/main/ui/workspace/sidebar/options/menus/exports.cljs:273
msgid "workspace.options.exporting-object"
msgstr "Exportando…"

7
render-wasm/src/main.rs
View File

@ -743,15 +743,14 @@ pub extern "C" fn is_image_cached(
}
/// Evicts least-recently-used images until the store retains at most
/// `max_mb` megabytes of image data. Called by the headless exporter between
/// `max_bytes` bytes of image data. Called by the headless exporter between
/// requests — never mid-render, so an image can't disappear under a running
/// export; evicted images are re-provisioned by later requests that need
/// them. Returns the number of evicted images.
#[no_mangle]
#[wasm_error]
pub extern "C" fn evict_images_to_budget(max_mb: u32) -> Result<u32> {
let max_bytes = (max_mb as usize) * 1024 * 1024;
let evicted = get_resources().images.evict_to_budget(max_bytes);
pub extern "C" fn evict_images_to_budget(max_bytes: u32) -> Result<u32> {
let evicted = get_resources().images.evict_to_budget(max_bytes as usize);
Ok(evicted as u32)
}