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

🔥 Delete the exporter's HTTP surface and the old job system

Browse Source 
The exporter is a consumer's process now, so everything that only the
HTTP surface held goes away with it: the server, the router and the
cookie auth; the export handlers (the runner up past their pieces
through `app.consumer.plan`, and `handlers/resources` keeps only the
local artifacts); the scheduler with its admission caps; the redis
job store, publish and cancel-sweeper; and with the store gone, the
redis engine too — the pollers own plain ioredis connections.

`app.jobs` is the local cancel arm only: a runtime record the claimed
job is registered under, the flag the parts of a run read between
renders, the shared-array signal a render thread reads inside a render,
the terminate callbacks, and a `mark-cancelled` that marks the record
terminal before the callbacks run. The row of the job was never this
process's business.

The pools size themselves to the worker concurrency, which is what a
process of K pollers can run at once; there is no independent pool cap
nor scheduler admission left. Docker: the nginx export proxy and the
exporter uri die with the endpoints they pointed at; the compose keeps
the exporter env thin and honest, the README describes the worker.

AI-assisted-by: glm-5.3-flash
This commit is contained in:
Andrey Antukh 2026-10-07 11:34:10 +00:00
parent 3a6ee0003f
commit b7ceb561f9
32 changed files with 338 additions and 2028 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1,47 +1,49 @@
# Exporter Architecture and Workflow
`exporter/`: CLJS/Node headless export service. Depends on `common/`; uses Playwright plus export JS/CLJS deps for SVG/PDF/assets.
`exporter/`: CLJS/Node **worker process** that consumes the backend's
`:exporter` job queue and renders shapes, frames and files to bitmap, SVG and
PDF. Depends on `common/`; uses Playwright plus export JS/CLJS deps for
SVG/PDF/assets. There is no HTTP surface of its own and nothing is persisted
on this side: the row of the job, its cancel and its result are the backend's.
## Layout and commands
- Source: `exporter/src/`; config: `deps.edn`, `shadow-cljs.edn`, `package.json`; runtime helpers/assets: `vendor/`, `scripts/`.
- From `exporter/`: setup `./scripts/setup`; watch `pnpm run watch` or `pnpm run watch:app`; production build `pnpm run build`; test bundle `pnpm run build:test`; tests `pnpm run test` or `pnpm run test:quiet`; lint `pnpm run lint:clj`; format check/fix `pnpm run check-fmt:clj` / `pnpm run fmt:clj`.
- From `exporter/`: setup `./scripts/setup`; watch `pnpm run watch` or `pnpm run watch:app`; production build `pnpm run build`; test bundle `pnpm run build:test`; tests `pnpm run test` or `pnpm run test:quiet`; lint `pnpm run lint:clj`; format check/fix `pnpm run check-fmt:clj` / `pnpm run fmt:clj`. `./scripts/ci exporter` in the repo root is green (test-quiet defaults `PENPOT_SECRET_KEY` like `test` does).
- Because exporter consumes `common/`, shared file/shape/model changes may need exporter verification even when the immediate change is not under `exporter/`.
- Cross-cutting testing principles and anti-patterns: `mem:testing`.
- Exporter test conventions and CI: `mem:exporter/testing`.
- Promise chains in the exporter (management client, auth) follow promesa's two API families; arg orders, what each fn must return, the `->>` preference and the async-test silent-green trap: `mem:clojure/promesa` — read before touching promise code.
- Promise chains in the exporter (management client, runner) follow promesa's two API families; arg orders, what each fn must return, the `->>` preference and the async-test silent-green trap: `mem:clojure/promesa` — read before touching promise code.
## HTTP and browser pool (legacy surface, doomed)
## The worker anatomy
- POST body limit is about 60 MB. Exporter supports `application/transit+json`; request params merge query params and body params.
- Map response bodies are Transit JSON and force HTTP 200; nil 200 bodies become 204.
- Auth token comes from cookie `auth-token`, then uploads use Bearer auth plus the management shared key.
- Each export job gets a fresh Playwright browser context. On success, the context closes and the browser returns to the pool; on error, the browser is destroyed instead of reused.
- Borrow validates browser connection. Pool acquire timeout is about 10s; font loading timeout logs a warning and continues after about 15s.
- The `http` role needs `PENPOT_EXPORTER_ROLES=http`. Default is gone once the HTTP deletion task lands; the scheduler config keys (`PENPOT_EXPORTER_MAX_CONCURRENT_JOBS` and friends) belong to this surface.
## Worker (consumer of the jobs substrate)
- Roles: `PENPOT_EXPORTER_ROLES` selects `http` and/or `worker`; **default is `worker`** (consumer-only process, no HTTP server to preserve): `app.consumer.config`, all accessors read live config.
- `worker` boots K pollers (`app.consumer.worker`): one Redis connection per poller (connection = slot + backpressure), BLPOP on `penpot.worker.queue:<tenant>:exporter`, payload `[job-id scheduled-at]` JSON; corrupt payload warns and drops, dead Redis delays 1s and retries. `K = PENPOT_EXPORTER_WORKER_CONCURRENCY` default 2, floor 1; replicas × K is the total concurrency. Pollers start after pools in `core.cljs` and stop before redis in the shutdown.
- Main thread boots pools + K pollers (`core.cljs`): browser pool, wasm pool, temp-file cleaner (`jobs.utils/init`), then `consumer.worker/start!`. Shutdown: browser → wasm pool → pollers → done (the pollers own their connections; the pools unwind while an export is still in flight).
- K pollers (`app.consumer.worker`): one Redis connection per poller (connection = slot + backpressure), BLPOP on `penpot.worker.queue:<tenant>:exporter`, payload `[job-id scheduled-at]` JSON; corrupt payload warns and drops, dead Redis delays 1s and retries. `K = PENPOT_EXPORTER_WORKER_CONCURRENCY` default 2, floor 1; replicas × K is the total concurrency. Pools size themselves to K (`browser` and `wasm.pool` read `ccfg/concurrency`); there is no independent pool-max config anymore, only `:wasm-worker-pool-min`.
- Management client (`app.consumer.api`): claim/progress/create-job-session/complete-job (JSON and multipart via undici FormData)/fail-job, POST to `<internal-uri>/api/management/methods/<method>` with `X-Shared-Key: exporter <key>` (key = `PENPOT_EXPORTER_SHARED_KEY` or HKDF-derived from the secret); transit bodies, non-2xx errors carry `:status` and the error body.
- Runner (`app.consumer.exports`, `run-export!`): mints a render session (`create-job-session`), plans via `make-plan` (single → artifact of its type; multi → zip; frames → pdf via pdfunite), settles via `complete-job-with-artifact` multipart or `fail-job`; the settle never rejects.
- Runner (`app.consumer.exports`, `run-export!`): mints a render session (`create-job-session`), plans via `make-plan` (single → artifact of its type; multi → zip; frames → pdf via pdfunite), settles via `complete-job-with-artifact` multipart or `fail-job`; the settle never rejects. `app.consumer.plan` holds the render-plan pieces (name transducers, partition of 50, grouping by `[scale type]`).
- Beats: per-object milestones `:preparing/:rendering/:packaging` with `objects`/`pages` counters, throttled at 250ms; a watchdog repeats the last milestone every 1s.
- Cancellation: a `skip` answer on any beat means the row is terminal (cancelled, aborted, settled). The runner raises `:job-cancelled` between units of work; mid-Skia the watchdog fires the local port once (`jobs/mark-cancelled`: flag + SharedArrayBuffer + terminate callbacks registered by `renderer.wasm/with-scope`), killing the leased worker thread. The runner owns the local registry entry via `jobs/register!` and releases it on settle.
- The legacy job machinery (`app.jobs`, scheduler, store) only serves the `http` surfaces plus the local cancel arm; the worker reuses pools and `job.utils/track!/release!` for temp files.
## Export batching and async behavior
- `prepare-exports` groups entries by `[scale type]` and partitions groups into chunks of 50. Each partition uses file/page/share/name from its first item, so be careful if entries might cross those boundaries.
- Single-export response is used only when multiple export is not forced and there is exactly one prepared export containing exactly one object.
- Multi-object export can run async: when `wait` is false it returns a resource immediately and publishes progress/end/error to Redis by profile topic; when `wait` is true it waits for upload and returns the uploaded resource.
- Frame export returns a resource immediately and publishes Redis updates; it does not follow the same `wait` option path.
- ZIP entry names are sanitized and duplicates receive numeric suffixes.
- Cancellation: a `skip` answer on any beat means the row is terminal (cancelled, aborted, settled). The runner raises `:job-cancelled` between units of work; mid-Skia the watchdog fires the local port once (`jobs/mark-cancelled`: flag + terminal local record + SharedArrayBuffer + terminate callbacks registered by `renderer.wasm/with-scope`), killing the leased worker thread. The runner owns the local registry entry via `jobs/register!` and releases it on settle.
- `app.jobs` is the **local cancel arm only**: a runtime registry keyed by job-id (`register!`, `mark-cancelled`, `cancel-signal`, `on-cancel`, `cancelled?`, `release!`). `mark-cancelled` marks the local record terminal BEFORE the callbacks run and returns true/nil (marked-now / no-op for unowned or settled). No redis writes.
- Temp files: `app.jobs.utils/track!`/`release!` per job id; a boot-time clean drops what a previous process left (aged `:exporter-job-ttl`). After the deletion there is no redis-side store, cancel topic or abandoned-job sweep: an unclaimed-in-time row goes `aborted` by the backend's own lease GC.
## Render details
- Headless engines (wasm/Skia) lease one render worker for the whole run (`rd/with-scope`); browser renders go one DOM page per partition of 50.
- Each export gets a fresh Playwright browser context. On success the context closes and the browser returns to the pool; on error the browser is destroyed instead of reused. Borrow validates the connection; pool acquire timeout about 10s; font loading timeout logs a warning and continues after about 15s.
- Bitmap export differs for WASM vs non-WASM render paths: WASM forces Playwright `deviceScaleFactor` to 1 and passes scale through the render URL; non-WASM uses `deviceScaleFactor = scale`.
- WebP is produced by taking a PNG screenshot and converting it with ImageMagick.
- SVG export rasterizes text foreignObjects to PNG, converts through PPM/color masks/potrace, and reassembles SVG paths. It also replaces non-breaking spaces for SVG compatibility and drops empty defs/paths.
- PDF export injects `@page` sizing through raw browser `evaluate` JavaScript; that code cannot rely on CLJS runtime helpers.
- Temporary resources schedule local deletion, then uploads POST to `/api/management/methods/upload-tempfile` with `X-Shared-Key: exporter <management-key>` and Bearer auth.
- ZIP entry names are sanitized (`plan/sanitize-file-regex`) and duplicates receive numeric suffixes.
## On the render engines (do not lose in refactors)
- The wasm engines run on worker threads; the Skia wasm bundle needs
`../render-wasm/build export` to generate `src/app/wasm/shared.js` before
compiling.
- A render thread cannot read a flag: the cancel reaches it through the
SharedArrayBuffer signal, and a render that never answers is killed by the
terminate watchdog of `app.wasm.pool` (the same mechanism a hard cancel
rides on).
- `core.cljs` decides what boots by thread: the main thread consumes the
queue, a render worker (`wasm.worker/main`) renders.

1
.serena/memories/exporter/testing.md
View File

@ -12,5 +12,6 @@
- Focus a test var with `node target/tests/test.js --focus exporter-tests.renderer-svg-test/creates-the-correct-gradient-element`.
- Set app log level by appending `--log-level warn` (or `trace|debug|info|warn|error`).
- `test:quiet` accepts forwarded options but rebuilds the bundle; prefer the direct runner after `build:test` for focused runs.
- The direct runner and `test`/`test:quiet` default `PENPOT_SECRET_KEY=test-secret-key` when the environment says nothing: the config needs a secret to boot.
- From `exporter/`: `pnpm run check-fmt:clj` checks ClojureScript formatting.
- From `exporter/`: `pnpm run lint:clj` runs ClojureScript linting.

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

@ -211,6 +211,9 @@ services:
- penpot
penpot-exporter:
# The exporter is a consumer's process: it polls the backend's
# `:exporter` job queue through valkey and calls the backend's
# management API to claim, report and settle every export job.
image: "penpotapp/exporter:${PENPOT_VERSION:-2.18}"
restart: always
@ -225,17 +228,17 @@ services:
<< : [*penpot-flags, *penpot-secret-key, *penpot-public-uri]
# Don't touch it; this uses an internal docker network to
# communicate with the frontend.
# reach the backend's management API and the frontend of the
# assets the exports render.
PENPOT_INTERNAL_URI: http://penpot-frontend:8080
## Valkey (or previously Redis) is used for the websockets notifications
## and for storing the state export jobs
## Valkey (or previously Redis) carries the export jobs queue the
## backend dispatcher pushes to, plus the websockets
## notifications
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_EXPORTER_WORKER_CONCURRENCY: 2
# PENPOT_WASM_WORKER_POOL_MAX: 2
# PENPOT_WASM_WORKER_POOL_MIN: 1
# PENPOT_WASM_WORKER_IDLE_TIMEOUT: 300

3
docker/images/files/nginx-entrypoint.sh
View File

@ -52,13 +52,12 @@ update_oidc_name /var/www/app/js/config.js
#########################################
export PENPOT_BACKEND_URI=${PENPOT_BACKEND_URI:-http://penpot-backend:6060}
export PENPOT_EXPORTER_URI=${PENPOT_EXPORTER_URI:-http://penpot-exporter:6061}
export PENPOT_HTTP_SERVER_MAX_BODY_SIZE=${PENPOT_HTTP_SERVER_MAX_BODY_SIZE:-367001600} # Default to 350MiB
export PENPOT_IPV6_LISTEN_DIRECTIVE=${PENPOT_IPV6_LISTEN_DIRECTIVE:-"listen [::]:8080 default_server reuseport backlog=16384;"}
if is_truthy "${PENPOT_DISABLE_IPV6_LISTEN:-}"; then
export PENPOT_IPV6_LISTEN_DIRECTIVE=""
fi
envsubst "\$PENPOT_BACKEND_URI,\$PENPOT_EXPORTER_URI,\$PENPOT_HTTP_SERVER_MAX_BODY_SIZE,\$PENPOT_IPV6_LISTEN_DIRECTIVE" \
envsubst "\$PENPOT_BACKEND_URI,\$PENPOT_HTTP_SERVER_MAX_BODY_SIZE,\$PENPOT_IPV6_LISTEN_DIRECTIVE" \
< /tmp/nginx.conf.template > /etc/nginx/nginx.conf
if [[ $PENPOT_FLAGS == *"enable-admin-console"* ]]; then

5
docker/images/files/nginx.conf.template
View File

@ -148,11 +148,6 @@ http {
proxy_pass $PENPOT_BACKEND_URI/link-preview$is_args$args;
}
location /api/export {
proxy_set_header Host $proxy_host;
proxy_pass $PENPOT_EXPORTER_URI;
}
location /api {
proxy_set_header Host $proxy_host;
proxy_pass $PENPOT_BACKEND_URI/api;

134
exporter/README.md
View File

@ -1,99 +1,55 @@
# 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.
Node process that renders shapes, frames and files to bitmap, SVG and PDF. It
is a **consumer**: it polls the backend's `:exporter` job queue on Redis and
drives each export through the backend's management API — claim, report the
milestones, and settle (`complete-job` multipart with the artifact, or
`fail-job` on error). There is no HTTP surface of its own and nothing is
persisted on this side: the row of the job, its cancel button and its result
all live in the backend.
## HTTP API
## How a job runs
Mounted under `/api/export` (the router matches on the path *after* that prefix,
so it also works when the process is hit directly on `/`).
1. The backend dispatcher pushes `[job-id, scheduled-at]` (JSON) into
`penpot.worker.queue:<tenant>:exporter`.
2. One of the `K` pollers pops it and owns that connection for the whole run:
the blocking pop is both the semaphore of the slot and the backpressure, so
the process never runs more exports than pollers. `K` is
`PENPOT_EXPORTER_WORKER_CONCURRENCY` (default 2, and the browser and wasm
pools size themselves to the same figure).
3. The poller claims the job over the management API and, on `:run`, mints a
render session (`create-job-session`) through which every asset fetch
happens as the owner of the job.
4. The milestones are reported as beats (`:preparing`, `:rendering`,
`:packaging`, with the `objects` and `pages` counters). A watchdog re-sends
the last one every second, whatever the run is doing.
5. It settles with `complete-job` (the artifact rides multipart, the session
is closed in the same step) or `fail-job`. The settle never rejects: a
worker that cannot talk leaves the row to the backend's lease GC.
| 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 |
## 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.
A cancellation reaches the worker through the beats: a `skip` answer means the
backend no longer listens (the row is terminal), so the runner stops between
units of work; mid-Skia the watchdog terminates the leased render worker. The
row was already `cancelled` by the backend's `cancel-job`.
## 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 |
| Variable | Default | Description |
|------------------------------------------|---------|----------------------------------------------------|
| `PENPOT_PUBLIC_URI` | — | The frontend the assets render through |
| `PENPOT_INTERNAL_URI` | `public-uri` | Same network's frontend; the management API too |
| `PENPOT_REDIS_URI` | `redis://redis/0` | The queue and the rest of redis |
| `PENPOT_TENANT` | `default` | Queue prefix (workspace name in devenv) |
| `PENPOT_EXPORTER_SHARED_KEY` | — | The shared key; derived from the secret otherwise |
| `PENPOT_EXPORTER_JOB_TTL` | `3600` | Lifetime of the temp files a job owns, in seconds |
| `PENPOT_EXPORTER_WORKER_CONCURRENCY` | `2` | `K`: pollers, running exports, pool sizes |
| `PENPOT_WASM_WORKER_POOL_MIN` | `1` | Render 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.
## Inspecting the backend
Nothing about a job is stored here: `get-job` in the backend answers the full
row, and its `app.debug`/admin surfaces render the same records.

11
exporter/scripts/test-quiet.js
View File

@ -24,6 +24,13 @@ progress("Running tests...");
const result = spawnSync(
"node",
["target/tests/test.js", ...process.argv.slice(2)],
{ stdio: "inherit" },
);
{
stdio: "inherit",
// the config of the process needs a secret to even boot: same
// default the `test` script uses when the environment says nothing
env: {
...process.env,
PENPOT_SECRET_KEY: process.env.PENPOT_SECRET_KEY || "test-secret-key",
},
});
process.exit(result.status ?? 1);

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

@ -1,98 +0,0 @@
;; This Source Code Form is subject to the terms of the Mozilla Public
;; License, v. 2.0. If a copy of the MPL was not distributed with this
;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
;;
;; Copyright (c) KALEIDOS SUBSIDIARY SL
(ns app.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)

6
exporter/src/app/browser.cljs
View File

@ -11,7 +11,7 @@
["playwright" :as pw]
[app.common.exceptions :as ex]
[app.common.logging :as l]
[app.config :as cf]
[app.consumer.config :as ccfg]
[app.util.object :as obj]
[promesa.core :as p]))
@ -155,8 +155,8 @@
(defn init
[]
(let [opts #js {:max (cf/get :browser-pool-max 5)
:min (cf/get :browser-pool-min 0)
(let [opts #js {:max (max 1 (ccfg/concurrency))
:min 0
:testOnBorrow true
:evictionRunIntervalMillis 5000
:numTestsPerEvictionRun 5

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

@ -24,21 +24,12 @@
{:public-uri "http://localhost:3449"
;; :internal-uri nil ;; internal-uri cannot be nil
:tenant "default"
:host "localhost"
:http-server-port 6061
:http-server-host "0.0.0.0"
:tempdir "/tmp/penpot"
:redis-uri "redis://redis/0"
:exporter-max-concurrent-jobs 4
:exporter-max-jobs-per-profile 2
:exporter-queue-max 64
:exporter-job-ttl 3600
;; the roles this instance serves: by default the consumer of the
;; backend jobs queue; `http` keeps the legacy surfaces alive until
;; the deletion task takes them away
:exporter-roles "worker"
;; how many export jobs run at once: every poller owns one Redis
;; connection, and the pools size themselves to the same number
:exporter-worker-concurrency 2
:wasm-worker-pool-max 2
:wasm-worker-pool-min 1
:wasm-worker-idle-timeout 300
:wasm-worker-image-cache-size (* 128 1024 1024)})
@ -49,7 +40,6 @@
[:public-uri {:optional true} ::sm/uri]
[:internal-uri {:optional true} ::sm/uri]
[:exporter-shared-key {:optional true} :string]
[:host {:optional true} :string]
[:tenant {:optional true} :string]
[:flags {:optional true} [::sm/set :keyword]]
[:redis-uri {:optional true} :string]

40
exporter/src/app/consumer/config.cljs
View File

@ -5,50 +5,18 @@
;; Copyright (c) KALEIDOS SUBSIDIARY SL
(ns app.consumer.config
"The roles of the exporter process.
`PENPOT_EXPORTER_ROLES` names the roles the process serves: `http`
serves the surfaces the exporter used to serve (gone once the
deletion task lands), `worker` consumes the backend `:exporter`
queue. The default is `worker`: the exporter is a consumer's process,
and no surface of it needs to be preserved. The worker pieces never
read these values at import time: each accessor reads the live
configuration, so a long-lived process can be restarted with another
role without code changes."
"The knobs of the consumer: how many jobs run at once and which
queue they come from. Read live, so a long-lived process obeys a
configuration change on the next poll."
(:require
[app.config :as cf]
[cljs.core :as c]
[cuerdas.core :as str]))
[cljs.core :as c]))
(def queue-name
"The queue this worker consumes: the one the `:export-assets` job-def
routes its jobs to through `::jobs/queue-name`."
"exporter")
(defn- scan-roles
[spec]
(->> (if (string? spec)
(str/split spec ",")
spec)
(map str/trim)
(remove str/blank?)
(map keyword)
(into #{})))
(defn roles
"The set of roles this instance serves, as keywords."
([] (roles (cf/get :exporter-roles)))
([spec]
(if (nil? spec)
#{:worker}
(scan-roles spec))))
(defn worker-enabled?
"True when this instance also consumes the exporter queue."
([] (contains? (roles) :worker))
([roles]
(contains? roles :worker)))
(defn concurrency
"K pollers: each one owns its Redis connection, and the blocking
connection is the semaphore of the slot, so K is the whole concurrency

26
exporter/src/app/consumer/exports.cljs
View File

@ -7,17 +7,15 @@
(ns app.consumer.exports
"The runner of the `:export-assets` job.
Takes the params the job-def froze and transplant the legacy export
machinery into the jobs substrate: the name transducers and
partitioning of `app.handlers.export-shapes`, and the pdf assembly
its `export-frames` twin does, dress the same run the old `/api/export`
served. What changes around them is the frame: the session minted
through `create-job-session` renders as the user, the milestone
vocabulary of the progress contract dresses the beats (`:preparing`,
Runs one export end to end: the render plan of
`app.consumer.plan` (the name transducers and the partitioning that
the export surfaces always kept), the pdf assembly of the frames
run, the beats shaped by the progress contract (`:preparing`,
`:rendering`, `:packaging`, with the `objects` and `pages` counters
of every partition), and the settle is one `complete-job` multipart
call that stores the artifact and closes the session in the same
step.
of every partition) and one settle: a `complete-job` multipart call
that stores the artifact and closes the session the backend minted
in the same step. The render runs as the job's owner through the
session `create-job-session` opened.
Cancellation arrives through the beats: a `skip` answer from
`report-job-progress` says the job can no longer hear the worker
@ -34,7 +32,7 @@
[app.common.exceptions :as ex]
[app.common.logging :as l]
[app.consumer.api :as api]
[app.handlers.export-shapes :as shapes]
[app.consumer.plan :as plan]
[app.handlers.resources :as rsc]
[app.jobs :as jobs]
[app.jobs.utils :as job.utils]
@ -143,7 +141,7 @@
[token {:keys [exports force-multiple name skip-children is-wasm]}]
(let [items (normalize-items exports)
frames? (boolean (every? frame-item? items))
prepared (shapes/prepare-exports items token is-wasm)
prepared (plan/prepare-exports items token is-wasm)
single? (and (not frames?)
(not (true? force-multiple))
(= 1 (count prepared))
@ -155,7 +153,7 @@
{:frames? frames?
:single? single?
:prepared prepared
:total (shapes/count-objects prepared)
:total (plan/count-objects prepared)
:skip-children skip-children
:is-wasm (boolean is-wasm)
:counter-kind (if frames? :pages :objects)
@ -230,7 +228,7 @@
(counter counter-kind @rendered total))
(rsc/add-to-zip zip path
(str/replace filename
shapes/sanitize-file-regex
plan/sanitize-file-regex
"_")))]
(p/let [beat-one (beats! :rendering (counter counter-kind 0 total) :force? true)
_ (check-beat! beat-one)

89
exporter/src/app/consumer/plan.cljs Normal file
View File

@ -0,0 +1,89 @@
;; This Source Code Form is subject to the terms of the Mozilla Public
;; License, v. 2.0. If a copy of the MPL was not distributed with this
;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
;;
;; Copyright (c) KALEIDOS SUBSIDIARY SL
(ns app.consumer.plan
"The render plan of an assets run.
The items the backend froze in the job become the renders this
process runs: file names built once and kept free of duplicates, then
grouped by scale and type, one render per group on the headless
engines and one DOM page per partition of fifty on the browser
engines. `app.consumer.exports` runs whatever this namespace plans,
and this namespace only plans."
(:require
[app.common.data :as d]
[app.renderer :as rd]
[app.util.mime :as mime]
[cuerdas.core :as str]))
;; Regex to clean namefiles
(def sanitize-file-regex #"[\\/:*?\"<>|]")
(defn count-objects
[exports]
(reduce + 0 (map (comp count :objects) exports)))
(defn- assoc-file-name
"A transducer that assocs a candidate filename and avoid duplicates"
[]
(letfn [(find-candidate [params used]
(loop [index 0]
(let [candidate (str (:name params)
(:suffix params "")
(when (pos? index)
(str/concat "-" (inc index)))
(mime/get-extension (:type params)))]
(if (contains? used candidate)
(recur (inc index))
candidate))))]
(fn [rf]
(let [used (volatile! #{})]
(fn
([] (rf))
([result] (rf result))
([result params]
(let [candidate (find-candidate params @used)
params (assoc params :filename candidate)]
(vswap! used conj candidate)
(rf result params))))))))
(def ^:const ^:private
default-partition-size 50)
(defn prepare-exports
[exports token is-wasm]
(letfn [(process-group [[part1 :as group]]
;; Browser path renders one DOM page per partition; headless WASM
;; takes the whole group in one render.
(if (rd/headless? {:is-wasm is-wasm :type (:type part1)})
[(build-render group)]
(sequence (comp (partition-all default-partition-size)
(map build-render))
group)))
(build-render [[part1 :as part]]
{:file-id (:file-id part1)
:page-id (:page-id part1)
:share-id (:share-id part1)
:name (:name part1)
:token token
:type (:type part1)
:scale (:scale part1)
:objects (mapv part-entry->object part)})
(part-entry->object [entry]
{:id (:object-id entry)
:filename (:filename entry)
:name (:name entry)
:suffix (:suffix entry)})]
(let [xform (comp
(map #(assoc % :token token))
(assoc-file-name))]
(->> (sequence xform exports)
(d/group-by (juxt :scale :type))
(map second)
(into [] (mapcat process-group))))))

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

@ -13,33 +13,19 @@
[app.config :as cf]
[app.consumer.config :as ccfg]
[app.consumer.worker :as worker]
[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]
[cuerdas.core :as str]
[promesa.core :as p]))
(enable-console-print!)
(l/setup! {:app :info})
(defn- roles-log
[]
(->> (ccfg/roles)
(map name)
(str/join ",")))
(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.
In the main thread the roles of the process decide the rest: `http`
boots the server and its pools exactly as always, `worker` boots the
consumer of the backend queue, and both can coexist during the
transition (the engines and the pools are shared)."
one render worker, or the single role of the main thread: consuming the
backend's export queue."
[& _]
(if-not ^boolean wt/isMainThread
(wasm.worker/main)
@ -47,25 +33,16 @@
(l/info :msg "initializing"
:public-uri (str (cf/get :public-uri))
:internal-uri (str (cf/get-internal-uri))
:roles (roles-log)
:workers (ccfg/concurrency)
:version (:full cf/version))
(l/info :msg "headless wasm export available"
: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)
(when (ccfg/worker-enabled?)
(worker/start!))
(when (contains? (ccfg/roles) :http)
(http/init))
;; With no role at all, the process boots into a no-op: the log
;; above is the only surface of that mistake, and honest.
nil))))
(worker/start!)))))
(def main start)
@ -99,13 +76,10 @@
(p/do
(shutdown-step "browser-pool" bwr/stop)
(shutdown-step "wasm-worker-pool" wasm.pool/stop)
;; the worker's pollers are inside their redis connections; ending
;; redis first would leave the pops on broken connections
(when (ccfg/worker-enabled?)
(shutdown-step "consumer-worker" worker/stop!))
(shutdown-step "redis" redis/stop)
(when (contains? (ccfg/roles) :http)
(shutdown-step "http" http/stop))
;; the pollers own their connections and may sit in a blocking
;; pop for the whole poll timeout; draining the pools first lets
;; the exports in flight unwind against their own resources
(shutdown-step "consumer-worker" worker/stop!)
(done)))))
(.on proc/default "uncaughtException"

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

@ -1,110 +0,0 @@
;; This Source Code Form is subject to the terms of the Mozilla Public
;; License, v. 2.0. If a copy of the MPL was not distributed with this
;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
;;
;; Copyright (c) KALEIDOS SUBSIDIARY SL
(ns app.handlers
(:require
[app.auth :as auth]
[app.common.data :as d]
[app.common.logging :as l]
[app.common.spec :as us]
[app.handlers.export :as export]
[app.util.transit :as t]
[clojure.spec.alpha :as s]
[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)
data (-> data
(assoc :explain explain)
(assoc :type :validation)
(dissoc ::s/problems ::s/value ::s/spec))]
(-> exchange
(assoc :response/status 400)
(assoc :response/body (t/encode data))
(assoc :response/headers {"content-type" "application/transit+json"})))
(= :not-found type)
(-> exchange
(assoc :response/status 404)
(assoc :response/body (t/encode data))
(assoc :response/headers {"content-type" "application/transit+json"}))
(and (= :internal type)
(= :browser-not-ready code))
(let [data {:type :server-error
:code :internal
:hint (ex-message error)
:data data}]
(-> exchange
(assoc :response/status 503)
(assoc :response/body (t/encode data))
(assoc :response/headers {"content-type" "application/transit+json"})))
:else
(let [data {:type :server-error
:code code
:hint (ex-message error)
:data data}]
(l/error :hint "unexpected internal error" :cause error)
(-> exchange
(assoc :response/status 500)
(assoc :response/body (t/encode (d/without-nils data)))
(assoc :response/headers {"content-type" "application/transit+json"}))))))
(defn handler
"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)
(->> (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))))))))))

100
exporter/src/app/handlers/export.cljs
View File

@ -1,100 +0,0 @@
;; This Source Code Form is subject to the terms of the Mozilla Public
;; License, v. 2.0. If a copy of the MPL was not distributed with this
;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
;;
;; Copyright (c) KALEIDOS SUBSIDIARY SL
(ns app.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
;; Backend used for admission and cancel UI: headless
;; WASM when `:is-wasm` is set, otherwise browser.
: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!))

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

@ -1,93 +0,0 @@
;; This Source Code Form is subject to the terms of the Mozilla Public
;; License, v. 2.0. If a copy of the MPL was not distributed with this
;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
;;
;; Copyright (c) KALEIDOS SUBSIDIARY SL
(ns app.handlers.export-frames
(:require
[app.common.spec :as us]
[app.handlers.export-shapes :refer [count-objects headless-exports? prepare-exports]]
[app.handlers.resources :as rsc]
[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 join-pdf)
(declare ^:private move-file)
(s/def ::name ::us/string)
(s/def ::file-id ::us/uuid)
(s/def ::page-id ::us/uuid)
(s/def ::object-id ::us/uuid)
(s/def ::is-wasm ::us/boolean)
(s/def ::export
(s/keys :req-un [::file-id ::page-id ::object-id ::name]))
(s/def ::exports
(s/every ::export :kind vector? :min-count 1))
(s/def ::params
(s/keys :req-un [::exports]
:opt-un [::name ::is-wasm]))
(defn- run-export
[job auth-token resource {:keys [exports is-wasm file-id]}]
(let [rendered (atom [])
on-object
(fn [{:keys [path] :as _object}]
(job.utils/track! (:id job) path)
(jobs/progress! job (count (swap! rendered conj path))))
exports
(map #(assoc % :is-wasm is-wasm :job-id (:id job)) exports)]
(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/fmap (fn [resource] (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
[job file-id paths]
(p/let [prefix (str/concat "penpot.pdfunite." file-id ".")
path (job.utils/track! (:id job) (sh/tempfile :prefix prefix :suffix ".pdf"))]
(apply sh/run-cmd! "pdfunite" (conj (vec paths) path))
path))
(defn- move-file
[{:keys [path] :as resource} output-path]
(p/do
(sh/move! output-path path)
resource))

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

@ -1,197 +0,0 @@
;; This Source Code Form is subject to the terms of the Mozilla Public
;; License, v. 2.0. If a copy of the MPL was not distributed with this
;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
;;
;; Copyright (c) KALEIDOS SUBSIDIARY SL
(ns app.handlers.export-shapes
(:require
[app.common.data :as d]
[app.common.spec :as us]
[app.handlers.resources :as rsc]
[app.jobs :as jobs]
[app.jobs.utils :as job.utils]
[app.renderer :as rd]
[app.util.mime :as mime]
[app.util.shell :as sh]
[cljs.spec.alpha :as s]
[cuerdas.core :as str]
[promesa.core :as p]))
(declare prepare-exports)
;; Regex to clean namefiles
(def sanitize-file-regex #"[\\/:*?\"<>|]")
(s/def ::file-id ::us/uuid)
(s/def ::filename ::us/string)
(s/def ::name ::us/string)
(s/def ::object-id ::us/uuid)
(s/def ::page-id ::us/uuid)
(s/def ::share-id ::us/uuid)
(s/def ::profile-id ::us/uuid)
(s/def ::scale ::us/number)
(s/def ::suffix ::us/string)
(s/def ::type ::us/keyword)
(s/def ::wait ::us/boolean)
(s/def ::is-wasm ::us/boolean)
(s/def ::export
(s/keys :req-un [::page-id ::file-id ::object-id ::type ::suffix ::scale ::name]
:opt-un [::share-id]))
(s/def ::exports
(s/coll-of ::export :kind vector? :min-count 1))
(s/def ::params
(s/keys :req-un [::exports ::profile-id]
:opt-un [::wait ::name ::skip-children ::force-multiple ::is-wasm]))
(defn count-objects
[exports]
(reduce + 0 (map (comp count :objects) exports)))
(defn- render!
[job export on-object]
(jobs/check-cancelled! job)
(rd/render (assoc export :job-id (:id job)) on-object))
(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))))))
(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)))
(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}))})
(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"
[]
(letfn [(find-candidate [params used]
(loop [index 0]
(let [candidate (str (:name params)
(:suffix params "")
(when (pos? index)
(str/concat "-" (inc index)))
(mime/get-extension (:type params)))]
(if (contains? used candidate)
(recur (inc index))
candidate))))]
(fn [rf]
(let [used (volatile! #{})]
(fn
([] (rf))
([result] (rf result))
([result params]
(let [candidate (find-candidate params @used)
params (assoc params :filename candidate)]
(vswap! used conj candidate)
(rf result params))))))))
(def ^:const ^:private
default-partition-size 50)
(defn prepare-exports
[exports token is-wasm]
(letfn [(process-group [[part1 :as group]]
;; Browser path renders one DOM page per partition; headless WASM
;; takes the whole group in one render.
(if (rd/headless? {:is-wasm is-wasm :type (:type part1)})
[(build-render group)]
(sequence (comp (partition-all default-partition-size)
(map build-render))
group)))
(build-render [[part1 :as part]]
{:file-id (:file-id part1)
:page-id (:page-id part1)
:share-id (:share-id part1)
:name (:name part1)
:token token
:type (:type part1)
:scale (:scale part1)
:objects (mapv part-entry->object part)})
(part-entry->object [entry]
{:id (:object-id entry)
:filename (:filename entry)
:name (:name entry)
:suffix (:suffix entry)})]
(let [xform (comp
(map #(assoc % :token token))
(assoc-file-name))]
(->> (sequence xform exports)
(d/group-by (juxt :scale :type))
(map second)
(into [] (mapcat process-group))))))

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

@ -1,60 +0,0 @@
;; This Source Code Form is subject to the terms of the Mozilla Public
;; License, v. 2.0. If a copy of the MPL was not distributed with this
;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
;;
;; Copyright (c) KALEIDOS SUBSIDIARY SL
(ns app.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"))))))

43
exporter/src/app/handlers/resources.cljs
View File

@ -5,18 +5,14 @@
;; Copyright (c) KALEIDOS SUBSIDIARY SL
(ns app.handlers.resources
"Temporal resources management."
"The run's local resources: the artifact files the runner parks in
the temp area until the settle moves them into the job's completion,
zipped once a run packs more than one."
(:require
["archiver" :as arc]
["node:fs" :as fs]
["node:fs/promises" :as fsp]
["node:path" :as path]
["undici" :as http]
[app.common.exceptions :as ex]
[app.common.transit :as t]
[app.common.uri :as u]
[app.common.uuid :as uuid]
[app.config :as cf]
[app.util.mime :as mime]
[app.util.shell :as sh]
[cljs.core :as c]
@ -63,36 +59,3 @@
(p/create (fn [resolve]
(.on ^js zip "close" resolve)
(.finalize ^js zip))))
(defn upload-resource
[auth-token resource]
(->> (fsp/readFile (:path resource))
(p/fmap (fn [buffer]
(new js/Blob #js [buffer] #js {:type (:mtype resource)})))
(p/mcat (fn [blob]
(let [fdata (new http/FormData)
agent (new http/Agent #js {:connect #js {:rejectUnauthorized false}})
headers #js {"X-Shared-Key" (str "exporter " cf/management-key)
"Authorization" (str "Bearer " auth-token)}
request #js {:headers headers
:method "POST"
:body fdata
:dispatcher agent}
uri (-> (cf/get-internal-uri)
(u/ensure-path-slash)
(u/join "api/management/methods/upload-tempfile")
(str))]
(.append fdata "content" blob (:filename resource))
(http/fetch uri request))))
(p/mcat (fn [response]
(if (not= (.-status response) 200)
(ex/raise :type :internal
:code :unable-to-upload-resource
:response-status (.-status response))
(.text response))))
(p/fmap t/decode-str)
(p/fmap (fn [result]
(merge resource (dissoc result :id))))))

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

@ -1,189 +0,0 @@
;; This Source Code Form is subject to the terms of the Mozilla Public
;; License, v. 2.0. If a copy of the MPL was not distributed with this
;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
;;
;; Copyright (c) KALEIDOS SUBSIDIARY SL
(ns app.http
(:require
["cookies$default" :as Cookies]
["inflation$default" :as inflate]
["node:http" :as http]
["node:stream$default" :as stream]
["raw-body$default" :as raw-body]
[app.common.logging :as l]
[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]))
(l/set-level! :info)
(defprotocol IStreamableResponseBody
(write-body! [_ response]))
(extend-protocol IStreamableResponseBody
string
(write-body! [data response]
(.write ^js response data)
(.end ^js response))
js/Buffer
(write-body! [data response]
(.write ^js response data)
(.end ^js response))
stream/Stream
(write-body! [data response]
(.pipe ^js data response)
(.on ^js data "error" (fn [cause]
(js/console.error cause)
(.end response)))))
(defn- handle-response
[{:keys [:response/body
:response/headers
:response/status
response]
:as exchange}]
(let [status (or status 200)
headers (clj->js headers)
body (or body "")]
(.writeHead ^js response status headers)
(write-body! body response)))
(defn- parse-headers
[req]
(let [orig (unchecked-get req "headers")]
(persistent!
(reduce #(assoc! %1 (str/lower %2) (unchecked-get orig %2))
(transient {})
(js/Object.keys orig)))))
(defn- wrap-body-params
[handler]
(let [opts #js {:limit "60mb" :encoding "utf8"}]
(fn [{:keys [:request/method :request/headers request] :as exchange}]
(let [ctype (get headers "content-type")]
(if (= method "post")
(-> (raw-body (inflate request) opts)
(p/then (fn [data]
(cond-> data
(= ctype "application/transit+json")
(t/decode-str))))
(p/then (fn [data]
(handler (assoc exchange :request/body-params data)))))
(handler exchange))))))
(defn- wrap-params
[handler]
(fn [{:keys [:request/body-params :request/query-params] :as exchange}]
(handler (assoc exchange :request/params (merge query-params body-params)))))
(defn- wrap-response-format
[handler]
(fn [exchange]
(p/then
(handler exchange)
(fn [{:keys [:response/body :response/status] :as exchange}]
(cond
(map? body)
(let [data (t/encode-str body {:type :json-verbose})
size (js/Buffer.byteLength data "utf-8")]
(-> exchange
(assoc :response/body data)
(assoc :response/status (or status 200))
(update :response/headers assoc "content-type" "application/transit+json")
(update :response/headers assoc "content-length" size)))
(and (nil? body)
(= 200 status))
(-> exchange
(assoc :response/body "")
(assoc :response/status 204)
(assoc :response/headers {"content-length" 0}))
:else
exchange)))))
(defn- wrap-query-params
[handler]
(fn [{:keys [:request/uri] :as exchange}]
(handler (assoc exchange :request/query-params (u/query-string->map (:query uri))))))
(defn- wrap-error
[handler on-error]
(fn [exchange]
(-> (p/do (handler exchange))
(p/catch (fn [cause] (on-error cause exchange))))))
(defn- wrap-auth
[handler cookie-name]
(fn [{:keys [:request/cookies] :as exchange}]
(let [token (.get ^js cookies cookie-name)]
(handler (cond-> exchange token (assoc :request/auth-token token))))))
(defn- wrap-health
"Add /healthz entry point intercept."
[handler]
(fn [{:keys [:request/path] :as exchange}]
(if (= path "/readyz")
(assoc exchange
:response/status 200
:response/body "OK")
(handler exchange))))
(defn- create-adapter
[handler]
(fn [req res]
(let [cookies (Cookies. req res)
headers (parse-headers req)
uri (u/uri (unchecked-get req "url"))
exchange {:request/method (str/lower (unchecked-get req "method"))
:request/path (:path uri)
:request/uri uri
:request/headers headers
:request/cookies cookies
:request req
:response res}]
(-> (p/do (handler exchange))
(p/then handle-response)))))
(defn- create-server
[handler]
(.createServer ^js http (create-adapter handler)))
(def instance (atom nil))
(defn init
[]
(let [handler (-> (router/create handlers/handler)
(wrap-health)
(wrap-auth "auth-token")
(wrap-response-format)
(wrap-params)
(wrap-query-params)
(wrap-body-params)
(wrap-error handlers/on-error))
server (create-server handler)
port (cf/get :http-server-port 6061)]
(.listen server port)
(l/info :hint "welcome to penpot"
:module "exporter"
:flags cf/flags
:version (:full cf/version))
(l/info :hint "starting http server" :port port)
(reset! instance server)))
(defn stop
[]
(if-let [server @instance]
(p/create (fn [resolve]
(.close server (fn []
(l/info :hint "shutdown http server")
(resolve)))))
(p/resolved nil)))

294
exporter/src/app/jobs.cljs
View File

@ -5,30 +5,17 @@
;; Copyright (c) KALEIDOS SUBSIDIARY SL
(ns app.jobs
"Export job model and lifecycle.
"The local arm of the job life this process runs.
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."
The durable row of a job is the backend's business: the worker claims
it, beats against it and settles it through the management API. This
namespace keeps only the runtime bits that cannot live in a row -- the
cancel callbacks that terminate a render worker, the cancel signal
shared with that render worker, and the flag the parts of a run read
between renders -- keyed by the job id of the claimed job."
(: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)
[app.common.time :as ct]))
(def ^:private terminal-states #{"ended" "error" "cancelled"})
@ -42,77 +29,16 @@
[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 cancelled?
"True when the job has been marked cancelled in this process. The
check between objects reads it; inside a render the thread reads the
signal instead."
[job-id]
(boolean (:cancelled? (runtime job-id))))
(defn cancel-signal
"Int32Array over a SharedArrayBuffer, readable from a worker thread: 0 while
@ -139,97 +65,11 @@
(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."
"Drops the local runtime entry once the job settled. The row stays in
the backend until it is read once more or expires."
[job-id]
(swap! registry dissoc (str job-id)))
(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 register!
"The runtime record of a job this process runs. The local cancel arm
reads it (`mark-cancelled`), so the in-flight render of a claimed job
@ -240,89 +80,35 @@
{:job {:id job-id :state "running" :total total}
:cancelled? false}))
(defn- check-cancel-callback!
[job-id f]
(try
(f)
(catch :default cause
(l/warn :hint "error on job cancel callback" :job-id (str job-id) :cause cause))))
(defn mark-cancelled
"Signals one job of this process as cancelled: raises the cancel flag
(the check between objects sees it), arms the shared-array signal (the
wasm worker sees it inside a render) and runs the cancel callbacks
(the terminate of the render worker). The durable row is already
terminal in the backend; no store write, no publish happens here.
(the check between objects sees it), marks the local record terminal
(the row of the job is already terminal in the backend), arms the
shared-array signal (the wasm worker sees it inside a render) and
runs the cancel callbacks (the terminate of the render worker). The
mark happens before the callbacks run: one of them releases the job,
and a second mark for the same job must change nothing.
A job this process does not own, or a settled one, is a no-op."
[job-id]
(let [k (str job-id)
rt (get @registry k)]
(when-let [rt (and rt (:job rt))]
(when-not (terminal? rt)
(swap! registry update k (fn [rt] (some-> rt (assoc :cancelled? true))))
(when-let [signal (:cancel-signal rt)]
(js/Atomics.store signal 0 1))
(doseq [f (:cancel-fns rt)]
(try
(f)
(catch :default cause
(l/warn :hint "error on job cancel callback" :job-id k :cause cause))))
(when-let [_job (and rt (:job rt))]
(when-not (terminal? (:job rt))
(let [callbacks (:cancel-fns rt)]
(swap! registry update k
(fn [e] (some-> e
(assoc :cancelled? true)
(update :job assoc :state "cancelled" :ended-at (now-ms)))))
(when-let [signal (:cancel-signal (get @registry k))]
(js/Atomics.store signal 0 1))
(doseq [f callbacks]
(check-cancel-callback! job-id f)))
true))))
(defn mark-cancelled!
"Marks a job of this process as cancelled and runs its cancel
callbacks (what terminates the render worker a hard-cancel needs) —
without touching the shared store.
It is the local arm of the manual `jobs/cancel-local!` has: the jobs
substrate marks the row `cancelled` in the backend, and the worker's
own records only need to know, so the in-flight render dies."
[job-id]
(cancel-local! job-id))
(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
View File

@ -1,149 +0,0 @@
;; This Source Code Form is subject to the terms of the Mozilla Public
;; License, v. 2.0. If a copy of the MPL was not distributed with this
;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
;;
;; Copyright (c) KALEIDOS SUBSIDIARY SL
(ns app.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
View File

@ -1,87 +0,0 @@
;; This Source Code Form is subject to the terms of the Mozilla Public
;; License, v. 2.0. If a copy of the MPL was not distributed with this
;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
;;
;; Copyright (c) KALEIDOS SUBSIDIARY SL
(ns app.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))

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

@ -1,177 +0,0 @@
;; This Source Code Form is subject to the terms of the Mozilla Public
;; License, v. 2.0. If a copy of the MPL was not distributed with this
;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
;;
;; Copyright (c) KALEIDOS SUBSIDIARY SL
(ns app.redis
(: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]
[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 role]
(let [^js client (new redis/default uri)]
(.on client "connect"
(fn [] (l/info :hint "redis connection established" :uri uri :role role)))
(.on client "error"
(fn [cause] (l/error :hint "error on redis connection" :role role :cause cause)))
(.on client "close"
(fn [] (l/warn :hint "connection closed" :role role)))
(.on client "reconnect"
(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" :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
[]
(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)]
(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" [])))

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

@ -1,61 +0,0 @@
;; This Source Code Form is subject to the terms of the Mozilla Public
;; License, v. 2.0. If a copy of the MPL was not distributed with this
;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
;;
;; Copyright (c) KALEIDOS SUBSIDIARY SL
(ns app.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))))))

18
exporter/src/app/wasm/pool.cljs
View File

@ -27,6 +27,7 @@
[app.common.logging :as l]
[app.common.transit :as t]
[app.config :as cf]
[app.consumer.config :as ccfg]
[promesa.core :as p]))
(l/set-level! :info)
@ -102,26 +103,23 @@
(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."
"How many renders can run at once: the pool holds as many render
workers as concurrent jobs the process runs, which is what the
pollers count. 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."
[]
;; 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)))
(max 1 (ccfg/concurrency)))
(defn init
[]
(let [configured (cf/get :wasm-worker-pool-max 2)
max-workers (capacity)
(let [max-workers (capacity)
opts #js {:max max-workers
:min (min max-workers (cf/get :wasm-worker-pool-min 1))
:testOnBorrow true
:evictionRunIntervalMillis 30000
:numTestsPerEvictionRun 2
:idleTimeoutMillis 300000}]
(when (not= configured max-workers)
(l/warn :hint "wasm-worker-pool-max raised to the minimum of one"
:configured configured))
(l/info :hint "initializing render worker pool" :opts opts)
(reset! pool (gp/createPool worker-pool-factory opts))
(p/resolved nil)))

24
exporter/test/exporter_tests/consumer_config_test.cljs
View File

@ -5,29 +5,11 @@
;; Copyright (c) KALEIDOS SUBSIDIARY SL
(ns exporter-tests.consumer-config-test
"The worker side of the config: the roles a process serves and how
many pollers a worker owns."
"The consumer side of the config: how many jobs run at once and
which queue they come from."
(:require
[app.consumer.config :as ccfg]
[cljs.test :as t :include-macros true]
[cuerdas.core :as str]))
(t/deftest roles-parse-the-env-list
(t/testing "the default serves the worker role only: the exporter is
a consumer's process, no surface needs preserving"
(t/is (= #{:worker} (ccfg/roles nil)))
(t/is (= #{:worker} (ccfg/roles))))
(t/is (= #{:http} (ccfg/roles "http")))
(t/is (= #{:http :worker} (ccfg/roles "http,worker")))
(t/is (= #{:http :worker} (ccfg/roles "http, worker")))
(t/testing "the parser does not invent roles from noise"
(t/is (= #{} (ccfg/roles "")))
(t/is (= #{:worker} (ccfg/roles ",worker")))))
(t/deftest worker-enabled-reads-the-roles
(t/is (true? (ccfg/worker-enabled? #{:http :worker})))
(t/is (true? (ccfg/worker-enabled? #{:worker})))
(t/is (false? (ccfg/worker-enabled? #{:http}))))
[cljs.test :as t :include-macros true]))
(t/deftest concurrency-is-at-least-one
(t/testing "the configured value is taken as given"

6
exporter/test/exporter_tests/export_shapes_test.cljs → exporter/test/exporter_tests/consumer_plan_test.cljs
View File

@ -4,11 +4,11 @@
;;
;; Copyright (c) KALEIDOS SUBSIDIARY SL
(ns exporter-tests.export-shapes-test
(ns exporter-tests.consumer-plan-test
"Chunking of the browser backend."
(:require
[app.common.uuid :as uuid]
[app.handlers.export-shapes :as export-shapes]
[app.consumer.plan :as plan]
[cljs.test :as t :include-macros true]))
(defn- exports
@ -26,6 +26,6 @@
(range n))))
(t/deftest browser-exports-are-chunked
(let [parts (export-shapes/prepare-exports (exports 120 :png 1) "token" false)]
(let [parts (plan/prepare-exports (exports 120 :png 1) "token" false)]
(t/is (= 3 (count parts)))
(t/is (= [50 50 20] (mapv (comp count :objects) parts)))))

131
exporter/test/exporter_tests/jobs_test.cljs
View File

@ -5,82 +5,85 @@
;; Copyright (c) KALEIDOS SUBSIDIARY 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."
"The local arm of the claimed job: the flag and the pieces a hard
cancel needs. The row of the job is the backend's business; nothing
here touches it."
(:require
[app.common.uuid :as uuid]
[app.jobs :as jobs]
[cljs.test :as t :include-macros true]
[promesa.core :as p]))
[cljs.test :as t :include-macros true]))
(defn- create!
(defn- register!
[]
(jobs/create! {:profile-id (uuid/next)
:cmd :export-shapes
:backend "wasm"
:total 10
:name "test"
:resource-id (uuid/next)}
(constantly (p/resolved nil))))
(let [job-id (uuid/next)]
(jobs/register! job-id)
job-id))
(t/deftest progress-does-not-resurrect-a-finished-job
(t/testing "a render reporting after the export failed cannot undo the failure"
(t/deftest mark-cancelled-runs-the-cancel-pieces-once
(t/testing "the signal arms, the callbacks run, and a second mark is a no-op"
(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))
(let [job-id (register!)
seen (atom [])]
(jobs/on-cancel job-id (fn [] (swap! seen conj :first)))
(jobs/on-cancel job-id (fn [] (swap! seen conj :second)))
(t/is (true? (jobs/mark-cancelled job-id)))
(t/is (true? (jobs/cancelled? job-id)))
(t/is (some? (jobs/cancel-signal job-id)))
;; the signal is armed: a render thread reading it stops
(let [^js signal (jobs/cancel-signal job-id)]
(t/is (= 1 (js/Atomics.load signal 0))))
(t/is (= [:first :second] @seen))
;; a settled job is a no-op: nothing marked, nobody new called
(t/is (nil? (jobs/mark-cancelled job-id)))
(t/is (= [:first :second] @seen))
(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 the-callbacks-see-the-cancelled-flag
;; a callback that stops the in-flight render reads the flag first: the
;; terminate of the worker is what a hard-cancel is for, and the pieces
;; below it must not race the writer
(t/async done
(let [job-id (register!)
seen (atom ::not-called)]
(jobs/on-cancel job-id (fn [] (reset! seen (jobs/cancelled? job-id))))
(jobs/mark-cancelled job-id)
(t/is (true? @seen))
(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/deftest a-signal-created-after-the-mark-is-armed
(t/testing "a job cancelled before its render leased a worker is
already stoppable for the render that comes next"
(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))))))
(let [job-id (register!)]
(jobs/mark-cancelled job-id)
(let [^js signal (jobs/cancel-signal job-id)]
(t/is (some? signal))
(t/is (= 1 (js/Atomics.load signal 0))))
(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/testing "a late cancel for a settled job changes nothing, and
release drops every piece"
(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))))
(let [job-id (register!)
seen (atom ::not-called)]
(jobs/on-cancel job-id (fn [] (reset! seen :callback-ran)))
(jobs/release! job-id)
(t/is (nil? (jobs/mark-cancelled job-id)))
(t/is (= ::not-called @seen))
(t/is (false? (jobs/cancelled? job-id)))
(t/is (nil? (jobs/cancel-signal job-id)))
(done)))))
(t/deftest an-unowned-job-is-a-no-op
(t/testing "mark and pieces for a job this process never registered"
(t/async done
(let [job-id (uuid/next)]
(t/is (nil? (jobs/mark-cancelled job-id)))
(t/is (false? (jobs/cancelled? job-id)))
(t/is (nil? (jobs/cancel-signal job-id)))
(done)))))

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

@ -12,10 +12,9 @@
[clojure.tools.cli :refer [parse-opts]]
[exporter-tests.api-test]
[exporter-tests.consumer-config-test]
[exporter-tests.export-shapes-test]
[exporter-tests.consumer-plan-test]
[exporter-tests.jobs-test]
[exporter-tests.renderer-svg-test]
[exporter-tests.scheduler-test]
[exporter-tests.shell-test]
[exporter-tests.wasm-pool-test]
[exporter-tests.worker-test]
@ -31,13 +30,12 @@
(println (str "UNHANDLED-REJECTION: " reason))))
(def test-namespaces
['exporter-tests.export-shapes-test
['exporter-tests.consumer-plan-test
'exporter-tests.jobs-test
'exporter-tests.api-test
'exporter-tests.worker-test
'exporter-tests.consumer-config-test
'exporter-tests.renderer-svg-test
'exporter-tests.scheduler-test
'exporter-tests.shell-test
'exporter-tests.wasm-pool-test])

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

@ -1,81 +0,0 @@
;; This Source Code Form is subject to the terms of the Mozilla Public
;; License, v. 2.0. If a copy of the MPL was not distributed with this
;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
;;
;; Copyright (c) KALEIDOS SUBSIDIARY SL
(ns exporter-tests.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))))))))